# This Python file uses the following encoding: utf-8 #from operator import inv #import os #from pathlib import Path from calendar import c import sys import time import random import struct import ctypes from ctypes.wintypes import HANDLE, UINT, WPARAM, LPARAM, DWORD, POINT # for hooking wndproc import win32api, win32con, win32gui from PySide6.QtWidgets import QApplication, QWidget, QDialog, QColorDialog, QLineEdit, QTextEdit, QGridLayout, QSizePolicy from PySide6.QtCore import QFile, QTimer, QThread, Signal, Slot, Qt from PySide6.QtGui import QColor, QPalette, QFont from PySide6.QtUiTools import QUiLoader from PySide6.QtCharts import QLineSeries, QChart import hid from hid import HIDException import numpy as np import pyaudio import wave import testgui_form USER_SIG_LENGTH = 512 # constants GWL_WNDPROC = -4 WM_DEVICECHANGE = 0x0219 DBT_DEVICEARRIVAL = 0x8000 DBT_DEVICEREMOVECOMPLETE = 0x8004 # Audio presets PREFERRED_API = 'WASAPI' MICROPHONE_NAME = 'Bigleap' PREFERRED_FORMAT = pyaudio.paInt16 PREFERRED_CHANNELS = 2 PREFERRED_RATE = 48000 PREFERRED_CHUNK = int(PREFERRED_RATE / 100) RECORDING_LENGTH = 5 #seconds # Command presets INTERCMD_DELAY = 0.1 # 100ms after each command sent INNERCMD_DELAY = 0.01 # 10ms for each command in a series class MSG(ctypes.Structure): ''' WinUser MSG - Contains message information from a thread's message queue. typedef struct tagMSG { HWND hwnd; UINT message; WPARAM wParam; LPARAM lParam; DWORD time; POINT pt; DWORD lPrivate; } MSG, *PMSG, *NPMSG, *LPMSG; ''' _pack_ = 1 _fields_ = [('hwnd', HANDLE), ('message', UINT), ('wparam', WPARAM), ('lparam', LPARAM), ('time', DWORD), ('pt', POINT), ('lPrivate', DWORD)] # transforms raw signature region data into a user signature structure: # currently it is: # struct st_user_sig_region{ # uint32_t magic_num; # char serial_number[16]; // 16 alpha-numeric serial number # uint16_t user_name[64]; // UTF-16 user name, up to 64 characters (will have to be truncated after 64, sorry long name folks) # RGB_Color_t default_color; // initial boot up color, and "off" (all zeros) can count as a color # uint8_t default_fan_speed; // fan speed in percent # bool linkbox_is_v1; // false (0): CC wire pulldown should be 1k (extra pulldowns ON) to assert HPD. # // true (1): CC wire pulldown should be 5.1k to assert HPD. This was the case on Linkbox v1. # uint8_t crc8; # } User_Signature_T; # # So byte-wise that is: # [0-3]: magic number (0x5AADE15F), LSB first # [4-19]: Serial number # [20-147]: user name # [148]: Red value # [149]: Green value # [150]: Blue value # [151]: Default fan speed # [152]: 1 if proximity sensor is disabled (doesn't control displays), 0 otherwise # [153]: 1 if we're using linkbox v1, 0 otherwise # [154]: crc8 - created by polynomial 0x07, no bit inversions, no bit reversals, initial value = 0xFF class sig_struct: MAGIC_NUM = 0x5AADE15F CRC_POLY = 0x07 CRC_INIT_VAL = 0xFF TOTAL_LENGTH = 154 def create_from_raw(self, rawdata): self.rawdata = rawdata # check that rawdata is 512 bytes long if(len(rawdata) != USER_SIG_LENGTH): print('bad length') self.valid = False else: # check magic number and crc magic_num = struct.unpack(' 16): serial_bytes = serial_bytes[0:16] if(len(serial_bytes) < 16): serial_bytes = serial_bytes + bytes([0]*(16-len(serial_bytes))) self.rawdata = self.rawdata + serial_bytes uname_bytes = uname.encode('utf-16-le') if(len(uname_bytes) > 128): uname_bytes = uname_bytes[0:128] if(len(uname_bytes) < 128): uname_bytes = uname_bytes + bytes([0]*(128-len(uname_bytes))) self.rawdata = self.rawdata + uname_bytes self.rawdata = self.rawdata + bytes([color.red(), color.green(), color.blue()]) self.rawdata = self.rawdata + bytes([fanspeed]) self.rawdata = self.rawdata + bytes([1 if prox_disable else 0]) self.rawdata = self.rawdata + bytes([1 if linkboxv1 else 0]) self.rawdata = self.rawdata + bytes([self.calc_crc8(self.rawdata)]) self.rawdata = self.rawdata + bytes([0xFF]*(USER_SIG_LENGTH - len(self.rawdata))) def calc_crc8(self, data_bytes): initval = self.CRC_INIT_VAL for bb in data_bytes: # for each byte, xor with the current value of CRC # then iterate over all 8 bits using the polynomial # to generate crc bits initval = initval ^ bb for _ in range(8): if(initval & 0x80): initval = ((initval << 1) & 0xFF) ^ self.CRC_POLY else: initval = (initval << 1) & 0xFF return initval class micreader(QThread): NUM_AVG_SAMPLES = 10 # About 1sec of audio to rolling average for level indicator mic_disconnected = Signal() recording_done = Signal() level_data = Signal(int) def __init__(self, mic_stream): self.astream = mic_stream super().__init__() def exit_now(self): self.quitting = True def save_wav(self, wave_filename, samplesize): self.saving_wave = True self.num_wave_saves = 0 self.wf = wave.open(wave_filename, 'wb') self.wf.setnchannels(PREFERRED_CHANNELS) self.wf.setsampwidth(samplesize) self.wf.setframerate(PREFERRED_RATE) self.wf.setnframes(RECORDING_LENGTH * PREFERRED_RATE) #wf.writeframes(b''.join(frames)) #wf.close() def run(self): self.quitting = False self.saving_wave = False rolling_average_left = 0 rolling_average_right = 0 average_place = 0 while(not self.quitting): try: chunk_data = self.astream.read(PREFERRED_CHUNK) # Convert bytes to signed ints chunk_ints = struct.unpack('<'+str(2*PREFERRED_CHUNK)+'h', chunk_data) # Stripe by channels, alternating samples are left and right channels left_ints = chunk_ints[::2] right_ints = chunk_ints[1::2] # Simple audio level - max (or min) value max_left = np.max(np.abs(left_ints)) max_right = np.max(np.abs(right_ints)) rolling_average_left = max_left + rolling_average_left rolling_average_right = max_right + rolling_average_right average_place = average_place + 1 if(average_place >= self.NUM_AVG_SAMPLES): average_place = 0 self.level_data.emit(int((rolling_average_left/self.NUM_AVG_SAMPLES) + (rolling_average_right/self.NUM_AVG_SAMPLES) / 2)) rolling_average_left = 0 rolling_average_right = 0 if(self.saving_wave): self.wf.writeframesraw(chunk_data) self.num_wave_saves = self.num_wave_saves + 1 if(self.num_wave_saves == int(PREFERRED_RATE / PREFERRED_CHUNK)*RECORDING_LENGTH): self.wf.close() self.saving_wave = False self.recording_done.emit() except(OSError): self.qutting = True self.mic_disconnected.emit() self.astream.close() class hidreader(QThread): HID_INPUT_REPORT_SIZE = 64 HID_FEATURE_REPORT_SIZE = 64 HID_TIMEOUT_MS = 10 data_received = Signal(bytes) hid_disconnected = Signal() def __init__(self, hid_device): self.dev = hid_device self.connected = True super().__init__() def exit_now(self): self.time_to_quit = True def write(self, wbytes): self.time_to_write = True self.stuff_to_write = wbytes def run(self): # periodically grab the input report from the connected HID self.time_to_quit = False self.time_to_write = False self.stuff_to_write = bytes([]) # infinite loop. read from the device with timeout while(not self.time_to_quit): try: result = self.dev.read(self.HID_INPUT_REPORT_SIZE, self.HID_TIMEOUT_MS) except(HIDException, OSError): self.hid_disconnected.emit() self.time_to_quit = True if(len(result) != 0): # We got something, send it to the UI thread self.data_received.emit(result) if(self.time_to_write): # got a push from the UI thread to write something to the HID self.time_to_write = False # stop after one write try: self.dev.send_feature_report(self.stuff_to_write) except(HIDException, OSError): self.hid_disconnected.emit() self.time_to_quit = True class RawSigPopUp(QDialog): def __init__(self): super().__init__() self.setWindowTitle("Raw Signature Data") self.editRawSig = QTextEdit(self) self.layout = QGridLayout() self.layout.addWidget(self.editRawSig, 1, 1, 1, 1) self.setLayout(self.layout) def setRawText(self, rawsigtext): self.editRawSig.setText("
"+rawsigtext+"
") #font = QFont() #font.setPointSize(8) #font.setKerning(True) #font.setFixedPitch(True) #font.setStyleHint(QFont.TypeWriter) #self.editRawSig.setFont(font) self.resize(800,600) #self.resize() class mainwin(QWidget): PROX_CHART_LENGTH = int(20) TEMP_CHART_LENGTH_TIME = int(3600) # 1 hour in seconds TEMP_CHART_PERIOD = int(5) # seconds between samples TEMP_CHART_LENGTH = int(TEMP_CHART_LENGTH_TIME / TEMP_CHART_PERIOD) def __init__(self): super(mainwin, self).__init__() self.current_device_list = [] self.is_hid_connected = False self.is_mic_connected = False self.hpd_enabled = True self.ssc_enabled = False self.status_text = 'Status: Disconnected' self.prev_brightness = 30 # default slider value self.rgb_color = QColor(0,0,0) self.label_palette = QPalette() self.label_palette.setColor(QPalette.Window, self.rgb_color) self.label_sig_palette = QPalette() self.label_sig_palette.setColor(QPalette.Window, self.rgb_color) # Override the WndProc just for the sole purpose of detecting when # a WM_DEVICECHANGE event occurs self.my_hwnd = self.winId() # print(self.my_hwnd) self.oldWndProc = win32gui.SetWindowLong(self.my_hwnd, win32con.GWL_WNDPROC, self.myWndProc) self.load_ui() self.link_actions() self.validate_vid_pid() self.create_prox_chart() self.create_temp_chart() self.refresh_hid_tree() def myWndProc(self, hWnd, msg, wParam, lParam): # print(msg, wParam, lParam) if(msg == win32con.WM_DEVICECHANGE): self.refresh_hid_tree() if(msg == win32con.WM_DESTROY): # restore the old wndproc for closing win32api.SetWindowLong(self.my_hwnd, win32con.GWL_WNDPROC, self.oldWndProc) return win32gui.CallWindowProc(self.oldWndProc, hWnd, msg, wParam, lParam) # Open the UI file and create the main window def load_ui(self): ''' loader = QUiLoader() path = os.fspath(Path(__file__).resolve().parent / 'form.ui') ui_file = QFile(path) ui_file.open(QFile.ReadOnly) self.widg = loader.load(ui_file, self) ui_file.close() ''' self.widg = testgui_form.Ui_mainwin() self.widg.setupUi(self) self.setWindowTitle('Bigscreen Displayboard Tester') self.widg.colorPickLabel.setPalette(self.label_palette) self.widg.lblSigColorView.setPalette(self.label_sig_palette) # Hide unused controls self.widg.btnEnableFan.setVisible(False) self.widg.btnEnableProx.setVisible(False) def resetStatus(self): self.widg.statusBar.setText(self.status_text) self.widg.statusBar.setStyleSheet('') # Connects actions to each button in the GUI def link_actions(self): self.widg.btnRefresh.clicked.connect(self.refresh_hid_tree) self.widg.btnConnect.clicked.connect(self.connect_hid) self.widg.editVID.textEdited.connect(self.validate_vid_pid) self.widg.editPID.textEdited.connect(self.validate_vid_pid) self.widg.sliderFanSpeed.valueChanged.connect(self.new_fan_speed) self.widg.btnEnableMic.clicked.connect(self.connect_mic) self.widg.btnSaveWavFile.clicked.connect(self.save_wav) self.widg.btnColorPicker.clicked.connect(self.launch_color_picker) self.widg.sliderBrightness.valueChanged.connect(self.update_brightness_pct) self.widg.sliderBrightness.sliderReleased.connect(self.update_brightness) self.widg.btnReadSig.clicked.connect(self.read_sig) self.widg.chkFlipDisplays.stateChanged.connect(self.flip_displays) self.widg.chkSwapDisplays.stateChanged.connect(self.swap_displays) self.widg.btnClearConsole.clicked.connect(self.clear_console) self.widg.btnSigSelColor.clicked.connect(self.launch_sig_color_picker) self.widg.btnSaveSig.clicked.connect(self.save_sig_to_flash) self.widg.btnSigViewRaw.clicked.connect(self.show_user_signature) self.widg.spinSigRed.valueChanged.connect(self.update_sig_color_from_spinner) self.widg.spinSigGreen.valueChanged.connect(self.update_sig_color_from_spinner) self.widg.spinSigBlue.valueChanged.connect(self.update_sig_color_from_spinner) def validate_vid_pid(self): try: self.auto_vid = int(self.widg.editVID.text(), 16) self.widg.editVID.setStyleSheet('') except(ValueError): self.widg.editVID.setStyleSheet('QLineEdit{background: rgb(243, 127, 127)}') self.auto_vid = None try: self.auto_pid = int(self.widg.editPID.text(), 16) self.widg.editPID.setStyleSheet('') except(ValueError): self.widg.editPID.setStyleSheet('QLineEdit{background: rgb(243, 127, 127)}') self.auto_pid = None # Get the currently attached USB HID devices and refresh the list widget def refresh_hid_tree(self): # Save previous selected item idx_path = None selected_idx = None if(len(self.current_device_list) > 0): idx = self.widg.listUSBHID.currentRow() # -1 is returned when nothing is selected, but that's a valid python index if(idx != -1): idx_path = self.current_device_list[idx]['path'] all_devs = hid.enumerate() self.widg.listUSBHID.clear() for dev in all_devs: listitem = '{}, VID: 0x{:04X}, PID: 0x{:04X}'.format( dev['product_string'], dev['vendor_id'], dev['product_id']) self.widg.listUSBHID.addItem(listitem) if(idx_path is not None): if(idx_path == dev['path']): selected_idx = self.widg.listUSBHID.count()-1 self.current_device_list = all_devs if(selected_idx is not None): self.widg.listUSBHID.setCurrentRow(selected_idx) # check if recognized device attached and auto-connect enabled if(self.widg.chkAutoConnect.isChecked() and not self.is_hid_connected): # Validate VID / PID if(self.auto_vid is not None and self.auto_pid is not None): for dev in self.current_device_list: if(dev['vendor_id'] == self.auto_vid and dev['product_id'] == self.auto_pid): self.connect_hid_by_path(dev['path']) def connect_hid(self): if(self.is_hid_connected): # This is the disconnect button now self.hidthread.exit_now() while(self.hidthread.isRunning()): pass self.hid_device.close() self.setUSBconnected(False) else: # Valid index in list? idx = self.widg.listUSBHID.currentRow() if(idx != -1): if(len(self.current_device_list) > 0): the_device = self.current_device_list[idx] self.connect_hid_by_path(the_device['path']) # self.hid_device = hid.Device(path=the_device['path']) # self.hid_device.nonblocking = 1 def connect_hid_by_path(self, path): try: self.hid_device = hid.Device(path=path) # Connected! self.setUSBconnected(True) except(HIDException): self.set_temporary_status('Status: Unable to connect', invert_colors=True) self.setUSBconnected(False) def connect_mic(self): if(not self.is_mic_connected): self.pyaudio_mngr = pyaudio.PyAudio() # find the correct audio device apis = [] for i in range(self.pyaudio_mngr.get_host_api_count()): api_dict = self.pyaudio_mngr.get_host_api_info_by_index(i) apis.append(api_dict['name']) if(PREFERRED_API in api_dict['name']): preferred_api_index = i for i in range(self.pyaudio_mngr.get_device_count()): dev_dict = self.pyaudio_mngr.get_device_info_by_index(i) if(dev_dict['hostApi'] == preferred_api_index): if(MICROPHONE_NAME in dev_dict['name']): # We can connect to this device! self.audiostream = self.pyaudio_mngr.open(format=PREFERRED_FORMAT, channels=PREFERRED_CHANNELS, rate=PREFERRED_RATE, input=True, input_device_index=i, frames_per_buffer=PREFERRED_CHUNK) self.set_audio_connected(True) if(not self.is_mic_connected): self.set_temporary_status('Could not connect to microphone', invert_colors=True) else: self.set_audio_connected(False) self.pyaudio_mngr.terminate() def save_wav(self): wavesaved = False if(self.is_mic_connected): if(hasattr(self,'micthread')): if(self.micthread.isRunning()): self.micthread.save_wav(self.widg.editWavFileName.text(), self.pyaudio_mngr.get_sample_size(PREFERRED_FORMAT)) self.micthread.recording_done.connect(self.wave_save_complete) wavesaved = True if(wavesaved): self.set_temporary_status('Recording wave file to '+self.widg.editWavFileName.text()) self.widg.btnSaveWavFile.setEnabled(False) self.widg.editWavFileName.setEnabled(False) self.widg.btnSelectWavFile.setEnabled(False) else: self.set_temporary_status('Error saving wave file',invert_colors=True) def wave_save_complete(self): self.set_temporary_status('Recording finished.') self.widg.btnSaveWavFile.setEnabled(True) self.widg.editWavFileName.setEnabled(True) self.widg.btnSelectWavFile.setEnabled(True) def launch_color_picker(self): self.colorDialog = QColorDialog() self.colorDialog.currentColorChanged.connect(self.new_color) self.colorDialog.open() def launch_sig_color_picker(self): self.sigColorDialog = QColorDialog() self.sigColorDialog.currentColorChanged.connect(self.new_sig_color) self.sigColorDialog.open() def new_color(self): newcolor = self.colorDialog.currentColor() new_r = newcolor.red() new_g = newcolor.green() new_b = newcolor.blue() self.widg.redValueLabel.setText(str(new_r)) self.widg.greenValueLabel.setText(str(new_g)) self.widg.blueValueLabel.setText(str(new_b)) if(hasattr(self,'hidthread') and self.hidthread.isRunning()): self.hidthread.write(bytes([0, ord('L'),new_r, new_g, new_b])) else: pass #print('New color: Red {}, Green {}, Blue {}'.format(new_r, new_g, new_b)) self.rgb_color = newcolor self.label_palette.setColor(QPalette.Window, self.rgb_color) self.widg.colorPickLabel.setPalette(self.label_palette) def new_sig_color(self): newcolor = self.sigColorDialog.currentColor() self.widg.spinSigRed.setValue(newcolor.red()) self.widg.spinSigGreen.setValue(newcolor.green()) self.widg.spinSigBlue.setValue(newcolor.blue()) self.sig_rgb_color = newcolor self.label_sig_palette.setColor(QPalette.Window, self.sig_rgb_color) self.widg.lblSigColorView.setPalette(self.label_sig_palette) def update_sig_color_from_spinner(self): self.sig_rgb_color = QColor(self.widg.spinSigRed.value(), self.widg.spinSigGreen.value(),self.widg.spinSigBlue.value()) self.label_sig_palette.setColor(QPalette.Window, self.sig_rgb_color) self.widg.lblSigColorView.setPalette(self.label_sig_palette) def setUSBconnected(self, is_connected): if(is_connected): self.widg.btnConnect.setText('Disconnect') self.widg.btnRefresh.setDisabled(True) self.widg.listUSBHID.setDisabled(True) self.widg.chkAutoConnect.setDisabled(True) self.is_hid_connected = True self.status_text = 'Status: Connected to {}'.format( self.hid_device.product) # print('Status: Connected to {}'.format(self.hid_device.product)) if(not self.is_status_timer_active()): self.resetStatus() self.hidthread = hidreader(self.hid_device) self.hidthread.hid_disconnected.connect(self.device_disconnected) self.hidthread.data_received.connect(self.device_data_received) self.hidthread.start() # Boost the reply rate from the microcontroller to 100ms time.sleep(0.2) self.hidthread.write(bytes([0, ord('R'), 0, 100])) else: self.widg.btnConnect.setText('Connect') self.widg.btnRefresh.setDisabled(False) self.widg.listUSBHID.setDisabled(False) self.widg.chkAutoConnect.setDisabled(False) self.is_hid_connected = False self.status_text = 'Status: Disconnected' if(not self.is_status_timer_active()): self.resetStatus() self.hidthread.exit_now() def set_audio_connected(self, is_connected): if(is_connected): self.widg.btnEnableMic.setText('Disable Microphone') self.is_mic_connected = True self.micthread = micreader(self.audiostream) self.micthread.level_data.connect(self.newAudioLevel) self.micthread.start() else: self.widg.btnEnableMic.setText('Enable Microphone') self.micthread.exit_now() while(self.micthread.isRunning()): pass self.is_mic_connected = False def newAudioLevel(self, newlevel): self.widg.dispMicLevel.display(newlevel) def device_disconnected(self): # print('Device connection lost') self.set_temporary_status('Connection lost', invert_colors=True) self.setUSBconnected(False) def read_sig(self): self.current_sig_subpage = 0 self.user_sig_data = bytearray([0]*USER_SIG_LENGTH) if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.set_temporary_status('Reading signature region {} of 16...'.format(self.current_sig_subpage)) self.hidthread.write(bytes([0, ord('U'), self.current_sig_subpage])) def show_user_signature(self): txtt = '' for i in range(int(USER_SIG_LENGTH/16)): txtt = txtt + '{:04X}: '.format(i*16) txtt = txtt + ''.join(['{:02X} '.format(bb) for bb in self.user_sig_data[16*i:(16*(i+1))]]) for bb in self.user_sig_data[16*i:(16*(i+1))]: thisbyte = bytes([bb]).decode('utf-8', errors='replace') if(thisbyte.isprintable()): txtt = txtt + thisbyte else: txtt = txtt + '.' txtt = txtt + '\n' siggypoppy = RawSigPopUp() siggypoppy.setRawText(txtt) siggypoppy.exec() def write_sig(self): pass ''' new_sig_bytes = 32*[0] sig_sub_block = self.widg.spinChooseSigBlock.value() try: for i in range(32): new_sig_bytes[i] = int(self.widg.sig_byte_box[i].text(), 16) if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.hidthread.write(bytes([0, ord('W'), sig_sub_block] + new_sig_bytes)) except(ValueError) as err: self.set_temporary_status('Invalid byte format in byte (use 00 through FF)'+'{0}'.format(err)) ''' def save_sig_to_flash(self): if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): my_sig_struct = sig_struct() my_sig_struct.create_from_parts( serial = self.widg.editSigSerialNum.text(), uname = self.widg.editSigUName.text(), fanspeed = self.widg.spinSigFanSpeed.value(), color = QColor(self.widg.spinSigRed.value(), self.widg.spinSigGreen.value(), self.widg.spinSigBlue.value()), prox_disable = self.widg.chkSigProxDisable.isChecked(), linkboxv1 = self.widg.chkSigLinkboxV1.isChecked() ) self.user_sig_data = my_sig_struct.rawdata # self.show_user_signature() self.current_sig_write_subpage = 0 new_sig_bytes = self.user_sig_data[self.current_sig_write_subpage*32:(self.current_sig_write_subpage+1)*32] self.hidthread.write(bytes([0, ord('W'), self.current_sig_write_subpage]) + new_sig_bytes) self.current_sig_write_subpage = self.current_sig_write_subpage + 1 QTimer.singleShot(10, self.save_next_sig_block) #self.hidthread.write(bytes([0, ord('V')])) def save_next_sig_block(self): if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): if(self.current_sig_write_subpage < (USER_SIG_LENGTH / 32)): # print('timer expired, next block is {}'.format(self.current_sig_write_subpage)) new_sig_bytes = self.user_sig_data[self.current_sig_write_subpage*32:(self.current_sig_write_subpage+1)*32] self.hidthread.write(bytes([0, ord('W'), self.current_sig_write_subpage]) + new_sig_bytes) self.current_sig_write_subpage = self.current_sig_write_subpage + 1 QTimer.singleShot(10, self.save_next_sig_block) else: # print('timer expired, now saving signature') self.hidthread.write(bytes([0, ord('V')])) def device_data_received(self, new_bytes): # print('got bytes: {}'.format(new_bytes)) # print('Packet type = {}, length = {}'.format(new_bytes[0], new_bytes[1])) # What type of data did we just receive: if((new_bytes[0] == ord('#')) and (new_bytes[1] == 20)): # Data packet (fan_speed, prox_value, cc1_value, cc2_value) = struct.unpack('>hhhh', new_bytes[2:(2 + 4*2)]) (board_temp, left_eye_temp, right_eye_temp) = struct.unpack('hhhh', new_bytes[12:20]) #print('Fan:{}, Prox:{}, cc1:{}, cc2:{}'.format(fan_speed, prox_value, cc1_value, cc2_value)) # print('Board temp:{}'.format(board_temp)) # print('Left eye temp:{}'.format(left_eye_temp)) # print('Right eye temp:{}'.format(right_eye_temp)) # print('Symbol Errors: Lane0={}, Lane1={}, Lane2={}, Lane3={}'.format(lane0symerr,lane1symerr,lane2symerr,lane3symerr)) self.widg.dispFanRPM.display(int(fan_speed)) self.widg.dispProxValue.display(int(prox_value)) self.widg.dispBoardTemp.display('{:.2f}'.format(board_temp - 273.15)) # board temp is in Kelvin self.widg.dispLeftEyeTemp.display('{:.2f}'.format(left_eye_temp)) # board temp is in Kelvin self.widg.dispRightEyeTemp.display('{:.2f}'.format(right_eye_temp)) # board temp is in Kelvin self.proxdata = self.proxdata[1:] + [prox_value] self.lineseries.removePoints(0, len(self.proxdata)) for i in range(len(self.proxdata)): self.lineseries.append(i, self.proxdata[i]) if(new_bytes[0] == ord('U')): # User signature reply packet user_sig_reply_len = new_bytes[1] user_sig_reply = new_bytes[2:(2+user_sig_reply_len)] # print("Got signature chunk #{}".format(self.current_sig_subpage)) # print(''.join(['0x{:02X}, '.format(bb) for bb in user_sig_reply])) # print('User signature subpage {}: {}'.format(self.current_sig_subpage, user_sig_reply.decode('utf-8'))) self.set_temporary_status('Reading signature region {} of 16...'.format(self.current_sig_subpage+1)) self.user_sig_data[32*self.current_sig_subpage:(32*(self.current_sig_subpage+1))] = user_sig_reply if(self.current_sig_subpage < 15): self.current_sig_subpage = self.current_sig_subpage + 1 self.hidthread.write(bytes([0, ord('U'), self.current_sig_subpage])) else: # check and save the returned signature my_sig_struct = sig_struct() my_sig_struct.create_from_raw(self.user_sig_data) if(my_sig_struct.valid): self.update_sig_fields(my_sig_struct) else: print('Invalid signature block') # self.show_user_signature() if(new_bytes[0] == ord('S')): # Debug printf packet # print(new_bytes[1:].decode('utf-8'),end='') txtt = self.widg.editConsole.toPlainText() txtt = txtt + new_bytes[1:].decode('utf-8') self.widg.editConsole.setPlainText(txtt.split('\x00',1)[0]) scrollbar = self.widg.editConsole.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) if(new_bytes[0] == ord('$')): # last command was successful # print('success') pass def update_sig_fields(self, newsig): self.widg.editSigSerialNum.setText(newsig.serial.split('\x00',1)[0]) self.widg.editSigUName.setText(newsig.username.split('\x00',1)[0]) self.widg.spinSigFanSpeed.setValue(newsig.fanspeed) self.widg.spinSigRed.setValue(newsig.color.red()) self.widg.spinSigGreen.setValue(newsig.color.green()) self.widg.spinSigBlue.setValue(newsig.color.blue()) self.update_sig_color_from_spinner() if(newsig.linkboxv1 == 1): self.widg.chkSigLinkboxV1.setChecked(True) else: self.widg.chkSigLinkboxV1.setChecked(False) if(newsig.prox_disable == 1): self.widg.chkSigProxDisable.setChecked(True) else: self.widg.chkSigProxDisable.setChecked(False) def set_temporary_status(self, status_text, invert_colors=False): self.widg.statusBar.setText(status_text) if(invert_colors): self.widg.statusBar.setStyleSheet('QLabel{background-color:rgb(0,0,0);color:rgb(255,255,255)}') if(self.is_status_timer_active()): self.status_bar_timer.stop() self.status_bar_timer = QTimer() self.status_bar_timer.timeout.connect(self.resetStatus) self.status_bar_timer.setSingleShot(True) self.status_bar_timer.start(2000) def is_status_timer_active(self): if(hasattr(self, 'status_bar_timer')): if(self.status_bar_timer.isActive()): return True return False def new_fan_speed(self, new_value): self.widg.dispFanPct.setText('{}%'.format(new_value)) if(self.is_hid_connected): # Send new fan speed command if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.hidthread.write(bytes([0, ord('F'), new_value])) def create_prox_chart(self): self.lineseries = QLineSeries() self.proxdata = [0]*self.PROX_CHART_LENGTH for i in range(len(self.proxdata)): self.lineseries.append(i, self.proxdata[i]) self.chart = QChart() self.chart.legend().hide() self.chart.addSeries(self.lineseries) self.chart.createDefaultAxes() self.chart.axes(orientation=Qt.Vertical)[0].setMax(17000) self.chart.axes(orientation=Qt.Vertical)[0].setMin(0) self.chart.setTitle('Proximity Sensor Data') self.widg.dispProxChart.setChart(self.chart) def create_temp_chart(self): self.boardtempseries = QLineSeries() self.boardtempseries.setName('Board Temp') self.boardtempdata = [0]*self.TEMP_CHART_LENGTH self.lefteyeseries = QLineSeries() self.lefteyeseries.setName('Left OLED Temp') self.lefteyedata = [0]*self.TEMP_CHART_LENGTH self.righteyeseries = QLineSeries() self.righteyeseries.setName('Right OLED Temp') self.righteyedata = [0]*self.TEMP_CHART_LENGTH for i in range(self.TEMP_CHART_LENGTH): self.boardtempseries.append(i*self.TEMP_CHART_PERIOD, self.boardtempdata[i]) self.lefteyeseries.append(i*self.TEMP_CHART_PERIOD, self.lefteyedata[i]) self.righteyeseries.append(i*self.TEMP_CHART_PERIOD, self.righteyedata[i]) self.tempchart = QChart() self.tempchart.addSeries(self.boardtempseries) self.tempchart.addSeries(self.lefteyeseries) self.tempchart.addSeries(self.righteyeseries) self.tempchart.createDefaultAxes() self.tempchart.axes(orientation=Qt.Vertical)[0].setMax(100) self.tempchart.axes(orientation=Qt.Vertical)[0].setMin(0) self.tempchart.setTitle('Temperature Data') self.widg.dispTempChart.setChart(self.tempchart) self.temperature_timer = QTimer() self.temperature_timer.timeout.connect(self.update_temp_chart) self.temperature_timer.start(self.TEMP_CHART_PERIOD * 1000) def update_temp_chart(self): # print('updating chart') self.boardtempdata = self.boardtempdata[1:] + [float(self.widg.dispBoardTemp.value())] self.lefteyedata = self.lefteyedata[1:] + [float(self.widg.dispLeftEyeTemp.value())] self.righteyedata = self.righteyedata[1:] + [float(self.widg.dispRightEyeTemp.value())] self.boardtempseries.removePoints(0, self.TEMP_CHART_LENGTH) self.lefteyeseries.removePoints(0, self.TEMP_CHART_LENGTH) self.righteyeseries.removePoints(0, self.TEMP_CHART_LENGTH) for i in range(self.TEMP_CHART_LENGTH): self.boardtempseries.append(i*self.TEMP_CHART_PERIOD, self.boardtempdata[i]) self.lefteyeseries.append(i*self.TEMP_CHART_PERIOD, self.lefteyedata[i]) self.righteyeseries.append(i*self.TEMP_CHART_PERIOD, self.righteyedata[i]) def send_ddic_cmd(self, i2c_addr, reg_addr, data): data_len = len(data) bytes_to_write = bytes([0, ord('D'),data_len,i2c_addr,reg_addr]) bytes_to_write = bytes_to_write + data if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.hidthread.write(bytes_to_write) return True # datastr = ''.join(['0x{:02X},'.format(bb) for bb in data]) # print('***send_ddic_cmd*** Dev: 0x{:02X}, Reg: 0x{:02X}, Data: {}'.format(i2c_addr, reg_addr, datastr)) return False def update_brightness_pct(self): # slider is "tracking", which means it will continually emit the "valueChanged" signal # while the user is sliding it # just update the value display, don't send the new brightness to the OLEDs until # slider is released self.widg.dispBrightPct.setText('{}%'.format(self.widg.sliderBrightness.value())) # just kidding, let's try live updates. self.update_brightness() #if(not self.widg.sliderBrightness.isSliderDown()): # Now update brightness. This can happen when the user clicks to the side of the slider # which causes it to page over without being pressed, or it can happen after clicking # and dragging the slider. #self.update_brightness() def update_brightness(self): # updates value when slider is released, but only if the value is different from the old value if(self.widg.sliderBrightness.value() != self.prev_brightness): self.prev_brightness = self.widg.sliderBrightness.value() if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.set_brightness(self.widg.sliderBrightness.value()) else: self.set_temporary_status('HID device not connected',True) else: self.set_temporary_status('HID device not connected',True) def set_brightness(self, new_brightness_pct): # requires three commands in a row: # reg (0xF0) / data (0xAA, 0x11) # reg (0xC0) / data (0x00, 0x40, 0x00) # reg (0xC2) / data (HB, LB, HB, LB, HB, LB, 0x00, 0x90, 0x02) # where HB is high byte of brightness, LB is low byte # value is 0x0000 to 0x03FF # all that is done on the MCU now new_brightness = int(new_brightness_pct * 1023 / 100) hb = (new_brightness & 0xFF00) >> 8 lb = new_brightness & 0xFF self.hidthread.write(bytes([0,ord('I'),hb,lb])) # set new brightness def flip_displays(self): this_succeeded = False if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): this_succeeded = True if(not self.widg.chkFlipDisplays.isChecked()): # Regular / unflipped: self.hidthread.write(bytes([0, ord('P'), 0])) else: # Flipped: self.hidthread.write(bytes([0, ord('P'), 1])) if(not this_succeeded): self.set_temporary_status('HID device not connected',True) self.widg.chkFlipDisplays.setChecked(False) def swap_displays(self): retval = False if(not self.widg.chkSwapDisplays.isChecked()): retval = self.send_ddic_cmd(0x10, 0x12, bytes([0x1B])) else: retval = self.send_ddic_cmd(0x10, 0x12, bytes([0xB1])) if(not retval): self.set_temporary_status('HID device not connected',True) self.widg.chkSwapDisplays.setChecked(False) def reset_DPRX(self): # Issues a DisplayPort Receiver soft reset. This sometimes fixes the issue # where screens quickly go pure white just after plugging in. if(hasattr(self,'hidthread')): if(self.hidthread.isRunning()): self.send_ddic_cmd(0x50, 0x04, bytes([0x08])) else: self.set_temporary_status('HID device not connected',True) else: self.set_temporary_status('HID device not connected',True) def send_long_mipi_cmd(self, mipi_reg, mipi_data, packet_type = 0x29): # first stage - load the payload fifo # must be multiples of 4 bytes fifo_dat = bytes([mipi_reg]) + mipi_data i = 0 while(i < len(fifo_dat)): bytes_to_write = bytes([0, ord('D'), 4, 0x18, 0x60]) if(i + 4 <= len(fifo_dat)): bytes_to_write = bytes_to_write + fifo_dat[i:i+4] else: zero_fill = i+4 - len(fifo_dat) bytes_to_write = bytes_to_write + fifo_dat[i:] + bytes([0]*zero_fill) self.hidthread.write(bytes_to_write) #print(''.join('0x{:02x},'.format(x) for x in bytes_to_write)) time.sleep(0.1) i = i + 4 # now send the mipi trigger command bytes_to_write = bytes([0, ord('D'), 3, 0x18, 0x5C, packet_type, len(fifo_dat)&0xFF, (len(fifo_dat)&0xFF00) >> 8]) self.hidthread.write(bytes_to_write) time.sleep(0.1) #print(''.join('0x{:02x},'.format(x) for x in bytes_to_write)) def clear_console(self): self.widg.editConsole.clear() # Overrides window native event, which can capture # device change notifications from the OS # This can be used to refresh the usb tree when a device is added/removed. '''def nativeEvent(self, eventType, message): # As far as I can tell, there is no intrinsic method to retrieve the # pointer address from a shiboken VoidPtr type # Thankfully, the string representation includes the address in its # text. The next line strips the address out of this text # Example: 'shiboken6.shiboken6.VoidPtr(Address 0x00000061BCFEACB0, Size 0, isWritable False)' address = int(message.__repr__().split('shiboken6.shiboken6.VoidPtr(')[1].split(',')[0], 16) newptr = ctypes.c_void_p(address) msg_struct_ptr = ctypes.cast(newptr, ctypes.POINTER(MSG)) msg_struct = msg_struct_ptr.contents # print('Message: {}, wParam: {}, lParam: {}'.format(msg_struct.message, msg_struct.wparam, msg_struct.lparam)) if(msg_struct.message == WM_DEVICECHANGE): # print('Status: New thing!') self.refresh_hid_tree() return super().nativeEvent(eventType, message)''' # Override close event to make sure the HID thread is safely stopped def closeEvent(self, event): if(hasattr(self, 'hidthread')): if(self.hidthread.isRunning()): self.hidthread.exit_now() while(self.hidthread.isRunning()): pass if(hasattr(self, 'micthread')): if(self.micthread.isRunning()): self.micthread.exit_now() while(self.micthread.isRunning()): pass return super().closeEvent(event) if __name__ == '__main__': app = QApplication([]) widget = mainwin() widget.show() sys.exit(app.exec())