import hid import struct import time import os import sys import enum from imgui_bundle import imgui, immapp from threading import Thread from queue import Queue import time class HIDReplies(enum.IntEnum): DATA = ord('#') TUNDRA_UART = ord(';') SUCCESS = ord('$') ERROR = ord('E') class HIDData: def __init__(self, fan:int, prox:int, cc1:int, cc2:int, board_temp:float, dispL_temp:float, dispR_temp:float, disp_bright:int, vid_status:int, vid_format:int, link_rate:int, lane_count:int, vxr_status:int): self.fan_speed = fan self.prox_distance = prox self.cc1 = cc1 self.cc2 = cc2 self.board_temp = board_temp self.dispL_temp = dispL_temp self.dispR_temp = dispR_temp self.disp_bright = disp_bright self.vid_status = vid_status self.vid_format = vid_format self.link_rate = link_rate self.lane_count = lane_count self.vxr_status = vxr_status class hid_receiver: def __init__(self, data_queue: Queue, tundra_uart_queue: Queue): self.connected = False self.quit_now = False self.data_queue = data_queue self.uart_queue = tundra_uart_queue self.hid_thread = None self.bynd = hid.device() self.connect() def connect(self): try: self.bynd.open(0x35BD, 0x0101) self.connected = True except: self.bynd.close() self.connected = False def stop(self): self.quit_now = True def start(self): self.hid_thread = Thread(target=self.receive_data) self.hid_thread.start() def decode_data(self, new_data: list[int]): if new_data[0] == HIDReplies.DATA: datalen = new_data[1] fan,prox,cc1,cc2 = struct.unpack('>HHHH',bytes(new_data[2:10])) board_t, dispL_t, dispR_t = struct.unpack('H', bytes(new_data[22:24])) video_status = new_data[25] & 0x0F video_format = (new_data[25] & 0xF0) >> 4 link_rate = new_data[24] & 0x0F lane_count = (new_data[24] & 0xF0) >> 4 vxr_status = new_data[26] & 0x03 if self.data_queue is not None: newdat = HIDData(fan, prox, cc1, cc2, board_t, dispL_t, dispR_t, disp_bright,video_status, video_format, link_rate, lane_count, vxr_status) self.data_queue.put(newdat) if new_data[0] == HIDReplies.TUNDRA_UART: datalen = new_data[1] newbytes = bytes(new_data[2:2+datalen]) self.uart_queue.put(newbytes.decode('utf8')) def receive_data(self): while not self.quit_now: if self.connected: try: next_data = self.bynd.read(65, timeout_ms = 10) except: self.connected = False next_data = [] self.bynd.close() if len(next_data) > 0: # try: self.decode_data(next_data) # print(next_data) # except: # pass else: try: time.sleep(0.25) self.bynd.open(0x35BD, 0x0101) self.connected = True except: # print("failed to connect") self.connected = False self.bynd.close() def label_and_value(label:str, value:str): imgui.table_next_column() imgui.text(label) imgui.table_next_column() imgui.text(value) imgui.table_next_row() class live_data_viewer: label_width = 400 def __init__(self): self.fan_speed = 0 self.prox_distance = 0 self.cc1_val = 0 self.cc2_val = 0 self.board_temp = 0.0 self.left_temp = 0.0 self.right_temp = 0.0 self.disp_brightness = 0 self.displays_on = False self.prox_on = False self.prox_timeout = False self.dsc = False self.vid_mode_string = "unknown" self.link_rate_string = "unknown" self.lane_count_string = "unknown" self.vxr_pwdn = True self.vxr_startup_good = False self.data_queue = Queue() self.tundra_uart_queue = Queue() self.hid_rx = hid_receiver(self.data_queue, self.tundra_uart_queue) self.autoscroll = False self.tundra_uart_buffer = "" def gui(self): scroll_down = False if self.hid_rx.hid_thread is None: self.hid_rx.start() if not self.data_queue.empty(): newdat = self.data_queue.get() if type(newdat) == HIDData: self.fan_speed = newdat.fan_speed self.prox_distance = newdat.prox_distance self.cc1_val = newdat.cc1 self.cc2_val = newdat.cc2 self.board_temp = newdat.board_temp self.left_temp = newdat.dispL_temp self.right_temp = newdat.dispR_temp self.disp_brightness = newdat.disp_bright self.displays_on = (newdat.vid_status & 0x01) != 0 self.prox_on = (newdat.vid_status & 0x02) != 0 self.prox_timeout = (newdat.vid_status & 0x04) != 0 self.dsc = (newdat.vid_status & 0x08) != 0 self.vid_mode_string = { 0: "unknown", 1: "H2544_V2544_75Hz", 2: "H2544_V2544_72Hz", 3: "H1920_V1920_90Hz", 4: "H1920_V1920_60Hz", 5: "H2560_V2560_30Hz", 6: "H2544_V2544_60Hz", 7: "Colorbar_Test"}[newdat.vid_format] self.link_rate_string = { 0: "unknown", 1: "RBR (1.62 Gbps)", 2: "HBR (2.7 Gbps)", 3: "HBR2 (5.4 Gbps)", 4: "HBR3 (8.1 Gbps)" }[newdat.link_rate] self.lane_count_string = str(newdat.lane_count) self.vxr_pwdn = (newdat.vxr_status & 0x01) != 0 self.vxr_startup_good = (newdat.vxr_status & 0x02) != 0 # elif type(newdat) == str: # self.tundra_uart_buffer += newdat # scroll_down = True if not self.tundra_uart_queue.empty(): newdat = self.tundra_uart_queue.get() self.tundra_uart_buffer += newdat scroll_down = True # Fan speed 2 bytes # Distance 2 bytes # CC1 2 bytes # CC2 2 bytes # Temp sensor (board) 4 bytes # Temp sensor Left oled 4 bytes # Temp sensor right oled 4 bytes # Brightness 2 bytes # Video Status 2 bytes # VXR status 1 byte if imgui.begin_table("vals_table", columns=2): imgui.table_next_row() label_and_value("Fan speed:" , f"{self.fan_speed}") label_and_value("Prox distance:" , f"{self.prox_distance}") label_and_value("USB-C CC1 value:" , f"{self.cc1_val}") label_and_value("USB-C CC2 value:" , f"{self.cc2_val}") label_and_value("Board temp:" , f"{self.board_temp}") label_and_value("Left oled temp:" , f"{self.left_temp}") label_and_value("Right oled temp:" , f"{self.right_temp}") label_and_value("Display brightness:" , f"{self.disp_brightness}") imgui.table_next_column() imgui.text(f"Displays on:") imgui.table_next_column() imgui.checkbox("##chk_displays_on", self.displays_on) imgui.table_next_row() imgui.table_next_column() imgui.text(f"Proximity on:") imgui.table_next_column() imgui.checkbox("##chk_prox_on", self.prox_on) imgui.table_next_row() imgui.table_next_column() imgui.text(f"Prox timer expired:") imgui.table_next_column() imgui.checkbox("##chk_prox_timeout", self.prox_timeout) imgui.table_next_row() imgui.table_next_column() imgui.text(f"DSC enabled:") imgui.table_next_column() imgui.checkbox("##chk_dsc", self.dsc) imgui.table_next_row() label_and_value(f"Video mode:" , f"{self.vid_mode_string}") label_and_value(f"DP Link Rate:" , f"{self.link_rate_string}") label_and_value(f"DP Lane Count:" , f"{self.lane_count_string}") imgui.table_next_column() imgui.text(f"VXR7200 powered down:") imgui.table_next_column() imgui.checkbox("##chk_vxr_pwdn", self.vxr_pwdn) imgui.table_next_row() imgui.table_next_column() imgui.text(f"VXR7200 startup success:") imgui.table_next_column() imgui.checkbox("##chk_vxr_startup_good", self.vxr_startup_good) imgui.table_next_row() imgui.end_table() imgui.begin_child("scroll_region", imgui.ImVec2(500, 200), imgui.ChildFlags_.none, imgui.WindowFlags_.always_vertical_scrollbar) imgui.text_unformatted(self.tundra_uart_buffer) if self.autoscroll and scroll_down: imgui.set_scroll_here_y(1.0) imgui.end_child() _, self.autoscroll = imgui.checkbox("Autoscroll", self.autoscroll) imgui.same_line() if imgui.button("Clear window?"): self.tundra_uart_buffer = "" if __name__ == "__main__": widget = live_data_viewer() immapp.run( gui_function = widget.gui, window_title="Bigscreen Live Data Viewer", # window_size_auto=True, window_size=(600,620) ) if widget.hid_rx.hid_thread.is_alive(): widget.hid_rx.stop() widget.hid_rx.hid_thread.join()