From a77da7554bdcfd8fd9fc9f4edc77e61485c3b876 Mon Sep 17 00:00:00 2001 From: IdkHowToCode123 Date: Mon, 23 Mar 2026 14:39:35 -0400 Subject: [PATCH 1/8] Syntax --- main.py | 192 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 main.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..dbcbb76 --- /dev/null +++ b/main.py @@ -0,0 +1,192 @@ +import dearpygui.dearpygui as dpg +import can +import threading +import queue + + +dpg.create_context() +bwid = 50 + +# --- THEMES --- +with dpg.theme() as green_led: + with dpg.theme_component(dpg.mvAll): + dpg.add_theme_color(dpg.mvThemeCol_Button, (34, 139, 34, 255)) + dpg.add_theme_style(dpg.mvStyleVar_FrameRounding, 50) + +with dpg.theme() as red_led: + with dpg.theme_component(dpg.mvAll): + dpg.add_theme_color(dpg.mvThemeCol_Button, (200, 0, 0, 255)) + dpg.add_theme_style(dpg.mvStyleVar_FrameRounding, 50) + +# --- UI LAYOUT --- +with dpg.window(label="Telemetry Master", tag="Primary Window"): + with dpg.group(horizontal=True): + with dpg.group(width=250): + with dpg.child_window(height=400, border=True): + dpg.add_text("Spark Maxes", color=(255, 255, 255)) + dpg.add_separator() + + with dpg.table(header_row=False, borders_innerH=False, borders_innerV=False): + for i in range(4): dpg.add_table_column() + + # DRIVE Section + with dpg.table_row(): dpg.add_text("DRIVE", color=(0, 255, 255)) + with dpg.table_row(): + # We give these specific tags so we can find them later + for i in range(1, 5): + tag_id = f"btn_sp{i}" #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER + dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) + dpg.bind_item_theme(tag_id, green_led) + + # ARM Section + with dpg.table_row(): dpg.add_text("ARM", color=(0, 255, 255)) + with dpg.table_row(): + for i in range(5, 8): #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER + tag_id = f"btn_sp{i}" + dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) + dpg.bind_item_theme(tag_id, green_led) + + #GRIP Section + with dpg.table_row(): dpg.add_text("GRIP", color=(0, 255, 255)) + with dpg.table_row(): + for i in range(9, 11): #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER + tag_id = f"btn_sp{i}" + dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) + dpg.bind_item_theme(tag_id, green_led) + + #SCIENCE SECTION + with dpg.table_row(): dpg.add_text("DRIVE", color=(0, 255, 255)) + with dpg.table_row(): + #REPLACE THIS ID WITH EXACT CAN ID VALUES LATER + tag_id="btn_sp19" + dpg.add_button(label="btn_sp19", width=bwid, height=bwid, tag=tag_id) + dpg.bind_item_theme(tag_id, green_led) + + with dpg.child_window(height=-1, border=True): + dpg.add_text("Science Module", color=(255, 255, 255)) + dpg.add_separator() + + with dpg.child_window(width=-1, border=True): + dpg.add_text("EBOX DATA", color=(0, 255, 255)) + dpg.add_separator() + +# --- CAN Bit Extractor --- +def extract_bits (bit, sbit_loc, bit_mask): + + return (bit >> sbit_loc) & bit_mask + +# --- THE LOGIC --- + +#FROM C Header, pass this to return as needed +''' + BITS 29 - 24 is type + '' 16 - 23 is manufacturer + '' 10-15 is class + '' 6 - 9 is index + and 0 - 5 is device id +''' +TYPE_POS = 24 +MANU_POS = 16 +CLASS_POS = 10 +INDEX_POS = 6 +DEVID_POS = 0 + +def get_telemetry_info(arb_id): + return { + "type": (arb_id >> TYPE_POS) & 0x1F, #5 bits + "manu": (arb_id >> MANU_POS) & 0xFF, #8 bits + "class": (arb_id >> CLASS_POS) & 0x3F, #6 bits + "index": (arb_id >> INDEX_POS) & 0x0F, #4 bits + "dev_id": (arb_id >> DEVID_POS) & 0x3F #6 bits + } + +#_ _ _ _ _ | _ _ _ _ _ _ _ | _ _ _ _ _ _ | _ _ _ _ | _ _ _ _ _ _ | +#type Menu Class Index Device ID + +#1. Create a queue + +telemetry_queue = queue.Queue() + +def can_worker(): + + try: + # Constants from C header - change when DBC is implemented + ''' + + + Creates a filter that ONLY lets through "Status 1" (where faults live) + This ignores the "Heartbeat" and "Encoder" data entirely at the hardware level. + + This ID represents: Type(02) + Mfr(05) + API(061) + DeviceID(00) + Binary: 0000 0010 + 0000 0101 + 0011 1101 + 0000 0000 + HEX: 0 2 0 5 3 D 0 0x + ''' + target_id = 0x02053D00 + + + #from CAN, we only want + filters = [ + { + "can_id": target_id, + "can_mask": 0x1FFFFFC0, # Check Type/Mfr/API, ignore Device ID + "extended": True} + ] + + bus = can.interface.Bus(channel='can0', bustype='socketcan', can_filters=filters) + except: + print("BUS NOT FOUND! (did you run sudo ip link...)?") + bus = None + while True: + # This BLOCKS. It stays here until a message arrives, instead of always polling + msg = bus.recv(timeout=1.0) + if msg: + info = get_telemetry_info(msg.arbitration_id) + #TELEMETRY: + if msg.arbitration_id == 0x02080546: #iD for Voltage / Current + print (f"VolCurr = {msg.data} from device {info["dev_id"]}") + if info["type"]== 2: + if info["manu"] == 8: + #for all sparkmaxes, 0x0208 or above + if info["index"] == 3: #temperature, for example + print(f"TEMP: {msg.arbitration_id} from device {info["dev_id"]}") + + #telemetry_queue.put((device_id, data)) # - gets passed into update_telemetry_ui + + + # 2. Start the thread before the UI loop +thread = threading.Thread(target=can_worker, daemon=True) +thread.start() + +# 3. Update the UI function to just check the Queue +def update_telemetry_ui(): + # Process everything currently in the queue + while not telemetry_queue.empty(): + try: + device_id, data = telemetry_queue.get_nowait() + target_tag = f"btn_sp{device_id}" #Going to have to change all the button values to the individual device IDs. + + if dpg.does_item_exist(target_tag): + if data > 0: #Data only passed if there is an issue + dpg.bind_item_theme(target_tag, red_led) + dpg.configure_item(target_tag, label=f"ID:{device_id}\nERR") + else: + dpg.bind_item_theme(target_tag, green_led) + dpg.configure_item(target_tag, label=f"SP{device_id}") + except queue.Empty: + break + + + + +# --- RENDER LOOP --- +dpg.create_viewport(title='YURS Telemetry', width=1200, height=800) +dpg.setup_dearpygui() +dpg.show_viewport() +dpg.set_primary_window("Primary Window", True) + +# Manual Render Loop (Crucial for live telemetry) +while dpg.is_dearpygui_running(): + update_telemetry_ui + dpg.render_dearpygui_frame() + +dpg.destroy_context() \ No newline at end of file From bac799c15a124f23c811d51bdf60abfa05ac36d1 Mon Sep 17 00:00:00 2001 From: IdkHowToCode123 Date: Mon, 30 Mar 2026 11:06:54 -0400 Subject: [PATCH 2/8] Changed to PySide6 --- Mac-CANTest.py | 49 +++++++ main.py | 347 ++++++++++++++++++++++------------------------- requirements.txt | 6 + 3 files changed, 217 insertions(+), 185 deletions(-) create mode 100644 Mac-CANTest.py create mode 100644 requirements.txt diff --git a/Mac-CANTest.py b/Mac-CANTest.py new file mode 100644 index 0000000..c39839e --- /dev/null +++ b/Mac-CANTest.py @@ -0,0 +1,49 @@ +import can +import time +import random + +def generate_fake_can(): + # Use 'virtual' for macOS compatibility + try: + bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast') + except: + # If vcan0 isn't defined in your configs, this usually just works + bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast') + + print("Bus started. Sending fake Rover telemetry...") + + # Constants from your whiteboard + TYPE = 2 + MANU = 8 + DEV_ID = 6 # Targeting Motor 6 as per your UI code + + count = 0 + while True: + # 1. Generate Heartbeat (Class 63, Index 0) + # ID Construction: (Type << 24) | (Manu << 16) | (Class << 10) | (Index << 6) | DevID + heartbeat_id = (TYPE << 24) | (MANU << 16) | (63 << 10) | (0 << 6) | DEV_ID + msg_hb = can.Message(arbitration_id=heartbeat_id, data=[1], is_extended_id=True) + bus.send(msg_hb) + #print(msg_hb) + + #another heartbeat + heartbeat_id = (TYPE << 24) | (MANU << 16) | (63 << 10) | (0 << 6) | 9 + msg_hb = can.Message(arbitration_id=heartbeat_id, data=[1], is_extended_id=True) + bus.send(msg_hb) + # 2. Generate Temp (Class 1, Index 3) + temp_id = (TYPE << 24) | (MANU << 16) | (1 << 10) | (3 << 6) | DEV_ID + fake_temp = int(40 + 10 * random.random()) # 40-50 degrees + msg_temp = can.Message(arbitration_id=temp_id, data=[fake_temp], is_extended_id=True) + bus.send(msg_temp) + + # 3. Generate Volt/Curr (Class 1, Index 4) + power_id = (TYPE << 24) | (MANU << 16) | (1 << 10) | (4 << 6) | DEV_ID + fake_volt = int(20 + random.random() * 4) # 20-24V + fake_curr = int(random.random() * 50) # 0-50A + msg_power = can.Message(arbitration_id=power_id, data=[fake_volt, fake_curr], is_extended_id=True) + bus.send(msg_power) + + time.sleep(0.1) # Send at 10Hz + +if __name__ == "__main__": + generate_fake_can() \ No newline at end of file diff --git a/main.py b/main.py index dbcbb76..7e4d7fd 100644 --- a/main.py +++ b/main.py @@ -1,192 +1,169 @@ -import dearpygui.dearpygui as dpg +import sys import can import threading -import queue - - -dpg.create_context() -bwid = 50 - -# --- THEMES --- -with dpg.theme() as green_led: - with dpg.theme_component(dpg.mvAll): - dpg.add_theme_color(dpg.mvThemeCol_Button, (34, 139, 34, 255)) - dpg.add_theme_style(dpg.mvStyleVar_FrameRounding, 50) - -with dpg.theme() as red_led: - with dpg.theme_component(dpg.mvAll): - dpg.add_theme_color(dpg.mvThemeCol_Button, (200, 0, 0, 255)) - dpg.add_theme_style(dpg.mvStyleVar_FrameRounding, 50) - -# --- UI LAYOUT --- -with dpg.window(label="Telemetry Master", tag="Primary Window"): - with dpg.group(horizontal=True): - with dpg.group(width=250): - with dpg.child_window(height=400, border=True): - dpg.add_text("Spark Maxes", color=(255, 255, 255)) - dpg.add_separator() - - with dpg.table(header_row=False, borders_innerH=False, borders_innerV=False): - for i in range(4): dpg.add_table_column() - - # DRIVE Section - with dpg.table_row(): dpg.add_text("DRIVE", color=(0, 255, 255)) - with dpg.table_row(): - # We give these specific tags so we can find them later - for i in range(1, 5): - tag_id = f"btn_sp{i}" #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER - dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) - dpg.bind_item_theme(tag_id, green_led) - - # ARM Section - with dpg.table_row(): dpg.add_text("ARM", color=(0, 255, 255)) - with dpg.table_row(): - for i in range(5, 8): #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER - tag_id = f"btn_sp{i}" - dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) - dpg.bind_item_theme(tag_id, green_led) - - #GRIP Section - with dpg.table_row(): dpg.add_text("GRIP", color=(0, 255, 255)) - with dpg.table_row(): - for i in range(9, 11): #REPLACE THIS FOR LOOP WITH EXACT CAN ID VALUES LATER - tag_id = f"btn_sp{i}" - dpg.add_button(label=f"SP{i}", width=bwid, height=bwid, tag=tag_id) - dpg.bind_item_theme(tag_id, green_led) - - #SCIENCE SECTION - with dpg.table_row(): dpg.add_text("DRIVE", color=(0, 255, 255)) - with dpg.table_row(): - #REPLACE THIS ID WITH EXACT CAN ID VALUES LATER - tag_id="btn_sp19" - dpg.add_button(label="btn_sp19", width=bwid, height=bwid, tag=tag_id) - dpg.bind_item_theme(tag_id, green_led) - - with dpg.child_window(height=-1, border=True): - dpg.add_text("Science Module", color=(255, 255, 255)) - dpg.add_separator() - - with dpg.child_window(width=-1, border=True): - dpg.add_text("EBOX DATA", color=(0, 255, 255)) - dpg.add_separator() - -# --- CAN Bit Extractor --- -def extract_bits (bit, sbit_loc, bit_mask): - - return (bit >> sbit_loc) & bit_mask - -# --- THE LOGIC --- - -#FROM C Header, pass this to return as needed -''' - BITS 29 - 24 is type - '' 16 - 23 is manufacturer - '' 10-15 is class - '' 6 - 9 is index - and 0 - 5 is device id -''' -TYPE_POS = 24 -MANU_POS = 16 -CLASS_POS = 10 -INDEX_POS = 6 -DEVID_POS = 0 - -def get_telemetry_info(arb_id): - return { - "type": (arb_id >> TYPE_POS) & 0x1F, #5 bits - "manu": (arb_id >> MANU_POS) & 0xFF, #8 bits - "class": (arb_id >> CLASS_POS) & 0x3F, #6 bits - "index": (arb_id >> INDEX_POS) & 0x0F, #4 bits - "dev_id": (arb_id >> DEVID_POS) & 0x3F #6 bits - } - -#_ _ _ _ _ | _ _ _ _ _ _ _ | _ _ _ _ _ _ | _ _ _ _ | _ _ _ _ _ _ | -#type Menu Class Index Device ID - -#1. Create a queue - -telemetry_queue = queue.Queue() - -def can_worker(): +#import struct #ONLY IF FLOATS ARE IN THE PAYLOAD +from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, + QHBoxLayout, QPushButton, QGridLayout) +from PySide6.QtCore import Qt, Signal, QObject +from PySide6.QtCharts import QChart, QChartView, QLineSeries, QValueAxis +from PySide6.QtGui import QColor # <--- The missing piece - try: - # Constants from C header - change when DBC is implemented - ''' +class CommWorker(QObject): + telemetry_received = Signal(int, dict, list) - - Creates a filter that ONLY lets through "Status 1" (where faults live) - This ignores the "Heartbeat" and "Encoder" data entirely at the hardware level. - - This ID represents: Type(02) + Mfr(05) + API(061) + DeviceID(00) - Binary: 0000 0010 + 0000 0101 + 0011 1101 + 0000 0000 - HEX: 0 2 0 5 3 D 0 0x - ''' - target_id = 0x02053D00 - - - #from CAN, we only want - filters = [ - { - "can_id": target_id, - "can_mask": 0x1FFFFFC0, # Check Type/Mfr/API, ignore Device ID - "extended": True} - ] - - bus = can.interface.Bus(channel='can0', bustype='socketcan', can_filters=filters) - except: - print("BUS NOT FOUND! (did you run sudo ip link...)?") - bus = None - while True: - # This BLOCKS. It stays here until a message arrives, instead of always polling - msg = bus.recv(timeout=1.0) - if msg: - info = get_telemetry_info(msg.arbitration_id) - #TELEMETRY: - if msg.arbitration_id == 0x02080546: #iD for Voltage / Current - print (f"VolCurr = {msg.data} from device {info["dev_id"]}") - if info["type"]== 2: - if info["manu"] == 8: - #for all sparkmaxes, 0x0208 or above - if info["index"] == 3: #temperature, for example - print(f"TEMP: {msg.arbitration_id} from device {info["dev_id"]}") - - #telemetry_queue.put((device_id, data)) # - gets passed into update_telemetry_ui - - - # 2. Start the thread before the UI loop -thread = threading.Thread(target=can_worker, daemon=True) -thread.start() - -# 3. Update the UI function to just check the Queue -def update_telemetry_ui(): - # Process everything currently in the queue - while not telemetry_queue.empty(): + def run(self): try: - device_id, data = telemetry_queue.get_nowait() - target_tag = f"btn_sp{device_id}" #Going to have to change all the button values to the individual device IDs. + # Filter logic: + # We want Class 1 (Telemetry) and Class 63 (Heartbeat). + # To keep it simple, we filter for Type=2, Manu=8. + # Mask 0x1F000000 checks Type, 0x00FF0000 checks Manu. + target_id = (2 << 24) | (8 << 16) + mask = 0x1FFF0000 - if dpg.does_item_exist(target_tag): - if data > 0: #Data only passed if there is an issue - dpg.bind_item_theme(target_tag, red_led) - dpg.configure_item(target_tag, label=f"ID:{device_id}\nERR") - else: - dpg.bind_item_theme(target_tag, green_led) - dpg.configure_item(target_tag, label=f"SP{device_id}") - except queue.Empty: - break - - + filters = [{"can_id": target_id, "can_mask": mask, "extended": True}] + #On mac, need to use UDP (dependency is removed from requirements.txt) + bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast',filter=filters) + except Exception as e: + print(f"CAN Bus Error: {e}") + return + + while True: + msg = bus.recv(timeout=1.0) + print(msg) + if msg: + # Bit mapping from your whiteboard + info = { + "dev_id": (msg.arbitration_id >> 0) & 0x3F, + "index": (msg.arbitration_id >> 6) & 0x0F, + "class": (msg.arbitration_id >> 10) & 0x3F, + "manu": (msg.arbitration_id >> 16) & 0xFF, + "type": (msg.arbitration_id >> 24) & 0x07, + } + self.telemetry_received.emit(msg.arbitration_id, info, list(msg.data)) + +class RoverDash(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("Sparky - Health Monitor") + self.resize(1024, 600) + self.setStyleSheet("background-color: #0f0f0f; color: white;") + + central_widget = QWidget() + self.setCentralWidget(central_widget) + main_layout = QHBoxLayout(central_widget) + + # --- LEFT SIDE: MOTOR STATUS --- + left_column = QVBoxLayout() + self.motor_grid = QGridLayout() + self.motor_buttons = {} + + for i in range(11): #replace this with the exact dev ids once you can get them + dev_id = i + 1 + btn = QPushButton(f"M{dev_id}") + btn.setFixedSize(60, 60) + btn.setStyleSheet(self.get_style("red")) # Default to red until heartbeat + self.motor_grid.addWidget(btn, i // 4, i % 4) + self.motor_buttons[dev_id] = btn + + btn12 = QPushButton("0x12") #e.g. button with dev id of 12 + btn12.setFixedSize(60,60) + btn12.setStyleSheet(self.get_style("red")) # Default to red until heartbeat + self.motor_grid.addWidget(btn12) + self.motor_buttons[12]=btn12 + + left_column.addLayout(self.motor_grid) + main_layout.addLayout(left_column, 1) + + # --- RIGHT SIDE: GRAPHS --- + graph_column = QVBoxLayout() + self.chart_grid = QGridLayout() + self.series_temp = QLineSeries() + self.chart_temp_view = self.create_graph("Temp (°C)", self.series_temp) + self.chart_grid.addWidget(self.chart_temp_view) + + self.series_current = QLineSeries() #cv represents current - voltage + self.series_current.setName("Current (A)") + self.series_current.setColor(QColor("#00e5ff")) # Cyan-ish for current + + self.series_voltage = QLineSeries() + self.series_voltage.setName("Voltage (V))") + self.series_voltage.setColor(QColor("#dc300e")) # red-ish for voltage + + self.chart_currVol_view = self.create_graph("Current & Voltage", [self.series_voltage, self.series_current]) + self.chart_grid.addWidget(self.chart_currVol_view) + + graph_column.addLayout(self.chart_grid) + main_layout.addLayout(graph_column, 2) + + self.data_count = 0 + + # --- WORKER THREAD --- + self.worker = CommWorker() + self.worker_thread = threading.Thread(target=self.worker.run, daemon=True) + self.worker.telemetry_received.connect(self.process_can_data) + self.worker_thread.start() + + def get_style(self, color): + hex_color = "#00ff88" if color == "green" else "#ff3333" + return f"background-color: {hex_color}; color: black; border-radius: 30px; font-weight: bold; border: 1px solid white;" + + def create_graph(self, title, series_input): + chart = QChart() + chart.setTitle(title) + chart.setTheme(QChart.ChartThemeDark) + + # Ensure we are working with a list even if one series is passed + series_list = series_input if isinstance(series_input, list) else [series_input] + + axis_x = QValueAxis() + axis_x.setRange(0, 100) + chart.addAxis(axis_x, Qt.AlignBottom) + + axis_y = QValueAxis() + axis_y.setRange(0, 100) # Adjust based on your battery/motor limits + chart.addAxis(axis_y, Qt.AlignLeft) + + for s in series_list: + chart.addSeries(s) + s.attachAxis(axis_x) + s.attachAxis(axis_y) + + view = QChartView(chart) + view.setRenderHint(view.renderHints().Antialiasing) + return view - -# --- RENDER LOOP --- -dpg.create_viewport(title='YURS Telemetry', width=1200, height=800) -dpg.setup_dearpygui() -dpg.show_viewport() -dpg.set_primary_window("Primary Window", True) - -# Manual Render Loop (Crucial for live telemetry) -while dpg.is_dearpygui_running(): - update_telemetry_ui - dpg.render_dearpygui_frame() - -dpg.destroy_context() \ No newline at end of file + def process_can_data(self, arb_id, info, data): + dev_id = info['dev_id'] + idx = info['index'] + cls = info['class'] + + # 1. Update Motor Status (Heartbeat) + if cls == 63: #heartbeat + if dev_id in self.motor_buttons: + self.motor_buttons[dev_id].setStyleSheet(self.get_style("green")) + + # 2. Update Graphs + if cls == 1: + if idx == 3 and dev_id == 6 and len(data) >= 1: # Temperature + # If data is a float, use: struct.unpack('=1: #current Voltage + vol=data[0] + amp=data[1] + self.series_current.append(self.data_count, vol) + self.series_voltage.append(self.data_count, amp) + + #Auto-scroll & Increment X-Axis + self.data_count += 1 + if self.data_count > 100: + new_min, new_max = self.data_count - 100, self.data_count + self.chart_temp_view.chart().axisX().setRange(new_min, new_max) + self.chart_currVol_view.chart().axisX().setRange(new_min, new_max) + +if __name__ == "__main__": + app = QApplication(sys.argv) + window = RoverDash() + window.show() + sys.exit(app.exec()) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6f07918 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +packaging==26.0 +PySide6==6.11.0 +PySide6_Addons==6.11.0 +PySide6_Essentials==6.11.0 +python-can==4.6.1 +shiboken6==6.11.0 From de571ffafde6c0d0c26818bcbe8da9ad44e1419e Mon Sep 17 00:00:00 2001 From: unknown Date: Tue, 9 Jun 2026 23:47:25 -0400 Subject: [PATCH 3/8] Streamlined, Commented, and Tested --- main.py | 199 +++++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 148 insertions(+), 51 deletions(-) diff --git a/main.py b/main.py index 7e4d7fd..e3dba9f 100644 --- a/main.py +++ b/main.py @@ -1,45 +1,74 @@ import sys import can -import threading -#import struct #ONLY IF FLOATS ARE IN THE PAYLOAD from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QGridLayout) -from PySide6.QtCore import Qt, Signal, QObject +from PySide6.QtCore import QThread, Qt, Signal, QObject, QTimer from PySide6.QtCharts import QChart, QChartView, QLineSeries, QValueAxis -from PySide6.QtGui import QColor # <--- The missing piece +from PySide6.QtGui import QColor +#this commworker handles all of the CAN communication. class CommWorker(QObject): - telemetry_received = Signal(int, dict, list) + error_status = Signal(int, int) + telemetry_received = Signal(dict, list) - def run(self): + def run(self): try: - # Filter logic: - # We want Class 1 (Telemetry) and Class 63 (Heartbeat). - # To keep it simple, we filter for Type=2, Manu=8. - # Mask 0x1F000000 checks Type, 0x00FF0000 checks Manu. - target_id = (2 << 24) | (8 << 16) - mask = 0x1FFF0000 - - filters = [{"can_id": target_id, "can_mask": mask, "extended": True}] - #On mac, need to use UDP (dependency is removed from requirements.txt) - bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast',filter=filters) + #Masks out only communication from SPARK devices. + target_id = (2 << 24) | (8 << 16) + mask = 0x1FFF0000 + + filters = [{ + "can_id": target_id, + "can_mask": mask, + "extended": True + }] + + bus = can.interface.Bus( + interface='slcan', + channel='/dev/tty.usbmodemXXXX', + bitrate=1000000, + filters=filters + ) except Exception as e: print(f"CAN Bus Error: {e}") return + self.running = True - while True: - msg = bus.recv(timeout=1.0) + while self.running: + msg = bus.recv(timeout=0.01) print(msg) - if msg: - # Bit mapping from your whiteboard - info = { - "dev_id": (msg.arbitration_id >> 0) & 0x3F, - "index": (msg.arbitration_id >> 6) & 0x0F, - "class": (msg.arbitration_id >> 10) & 0x3F, - "manu": (msg.arbitration_id >> 16) & 0xFF, - "type": (msg.arbitration_id >> 24) & 0x07, - } - self.telemetry_received.emit(msg.arbitration_id, info, list(msg.data)) + if msg is None: + continue + + can_id = msg.arbitration_id + info = { + "dev_id": (can_id >> 0) & 0x3F, #device ID of sparkmax [gets passed to btn] + "index": (can_id >> 6) & 0x0F, + "class": (can_id >> 10) & 0x3F, + "manu": (can_id >> 16) & 0xFF, + "type": (can_id >> 24) & 0x1F, + } + + if info["class"] == 61: + if info["index"] == 0: #error class is 61, index 0 (from ref sheet) + error_code = msg.data[0] #looks like the message sends error code (error code is mapped to LEDs in ref.) + self.error_status.emit(info["dev_id"], error_code) #sends values to update_led_status + self.telemetry_received.emit(info, list(msg.data)) + def stop(self): + self.running = False + + +LED_MAP = { + 1: ("red", "blue", "slow"), + 2: ("red", "cyan", "slow"), + 3: ("red", "green", "slow"), + 4: ("red", "magenta", "slow"), + 5: ("red", "yellow", "slow"), + 6: ("cyan", "cyan", "normal"), + 7: ("cyan", "cyan", "solid"), + 8: ("red", "cyan", "normal"), + 9: ("green", "green", "normal"), +} class RoverDash(QMainWindow): def __init__(self): @@ -52,25 +81,21 @@ def __init__(self): self.setCentralWidget(central_widget) main_layout = QHBoxLayout(central_widget) + # --- LEFT SIDE: MOTOR STATUS --- left_column = QVBoxLayout() self.motor_grid = QGridLayout() self.motor_buttons = {} - for i in range(11): #replace this with the exact dev ids once you can get them + for i in range(12): #create buttons btn1 - 12 dev_id = i + 1 - btn = QPushButton(f"M{dev_id}") + btn = QPushButton(f"axis{dev_id}") btn.setFixedSize(60, 60) - btn.setStyleSheet(self.get_style("red")) # Default to red until heartbeat + # btn.setStyleSheet(self.get_style("red")) # Default to red until heartbeat self.motor_grid.addWidget(btn, i // 4, i % 4) self.motor_buttons[dev_id] = btn - btn12 = QPushButton("0x12") #e.g. button with dev id of 12 - btn12.setFixedSize(60,60) - btn12.setStyleSheet(self.get_style("red")) # Default to red until heartbeat - self.motor_grid.addWidget(btn12) - self.motor_buttons[12]=btn12 - + left_column.addLayout(self.motor_grid) main_layout.addLayout(left_column, 1) @@ -78,7 +103,7 @@ def __init__(self): graph_column = QVBoxLayout() self.chart_grid = QGridLayout() self.series_temp = QLineSeries() - self.chart_temp_view = self.create_graph("Temp (°C)", self.series_temp) + self.chart_temp_view = self.create_graph("Temp (°C)", self.series_temp) self.chart_grid.addWidget(self.chart_temp_view) self.series_current = QLineSeries() #cv represents current - voltage @@ -99,14 +124,84 @@ def __init__(self): # --- WORKER THREAD --- self.worker = CommWorker() - self.worker_thread = threading.Thread(target=self.worker.run, daemon=True) + self.thread = QThread() + + self.worker.moveToThread(self.thread) + + self.thread.started.connect(self.worker.run) + + self.worker.error_status.connect(self.update_led_status) self.worker.telemetry_received.connect(self.process_can_data) - self.worker_thread.start() + + self.button_states = {} + self.blink_timers = {} + + self.thread.start() + + # --- LOGIC --- + + + def update_led_status(self, dev_id, error_code): + color1, color2, speed = LED_MAP.get(error_code, ("red", None, "solid")) + + btn = self.motor_buttons.get(dev_id) + if btn is None: + return + + if dev_id in self.blink_timers: + self.blink_timers[dev_id].stop() + del self.blink_timers[dev_id] + + if speed == "solid" or color2 is None: + btn.setStyleSheet(self.get_style(color1)) + return + + intervals = { + "normal": 250, + "slow": 500, + } + + interval = intervals.get(speed, 500) + + state = 1 + + def blink(): + nonlocal state + state = not state + + color = color1 if state else color2 + btn.setStyleSheet(self.get_style(color)) + + timer = QTimer(self) + timer.timeout.connect(blink) + timer.start(interval) + + self.blink_timers[dev_id] = timer + blink() def get_style(self, color): - hex_color = "#00ff88" if color == "green" else "#ff3333" - return f"background-color: {hex_color}; color: black; border-radius: 30px; font-weight: bold; border: 1px solid white;" + colors = { + "red": "#ff3333", + "green": "#00ff88", + "blue": "#3399ff", + "cyan": "#00e5ff", + "magenta": "#ff33cc", + "yellow": "#ffff33", + "off": "#222222", + } + hex_color = colors.get(color, "#ff3333") + + return f""" + QPushButton {{ + background-color: {hex_color}; + color: black; + border-radius: 8px; + font-weight: bold; + }} + """ + + def create_graph(self, title, series_input): chart = QChart() chart.setTitle(title) @@ -132,20 +227,15 @@ def create_graph(self, title, series_input): view.setRenderHint(view.renderHints().Antialiasing) return view - def process_can_data(self, arb_id, info, data): + + def process_can_data(self, info, data): + #WORK IN PROGRESS dev_id = info['dev_id'] idx = info['index'] cls = info['class'] - - # 1. Update Motor Status (Heartbeat) - if cls == 63: #heartbeat - if dev_id in self.motor_buttons: - self.motor_buttons[dev_id].setStyleSheet(self.get_style("green")) - - # 2. Update Graphs + if cls == 1: if idx == 3 and dev_id == 6 and len(data) >= 1: # Temperature - # If data is a float, use: struct.unpack(' Date: Tue, 23 Jun 2026 00:52:46 -0400 Subject: [PATCH 4/8] Removed Graph. Added text for voltage in volcurr payload. --- main.py | 189 ++++++++++++++++++++++++++++---------------------------- 1 file changed, 94 insertions(+), 95 deletions(-) diff --git a/main.py b/main.py index e3dba9f..a26ffdf 100644 --- a/main.py +++ b/main.py @@ -1,15 +1,13 @@ import sys import can -from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, - QHBoxLayout, QPushButton, QGridLayout) -from PySide6.QtCore import QThread, Qt, Signal, QObject, QTimer -from PySide6.QtCharts import QChart, QChartView, QLineSeries, QValueAxis -from PySide6.QtGui import QColor +from PySide6.QtWidgets import (QApplication, QLabel, QMainWindow, QWidget, QVBoxLayout, + QPushButton, QGridLayout) +from PySide6.QtCore import QThread, Signal, QObject, QTimer #this commworker handles all of the CAN communication. class CommWorker(QObject): error_status = Signal(int, int) - telemetry_received = Signal(dict, list) + volcurr = Signal(int,int) def run(self): try: @@ -25,7 +23,7 @@ def run(self): bus = can.interface.Bus( interface='slcan', - channel='/dev/tty.usbmodemXXXX', + channel='rover-test', bitrate=1000000, filters=filters ) @@ -48,12 +46,16 @@ def run(self): "manu": (can_id >> 16) & 0xFF, "type": (can_id >> 24) & 0x1F, } - - if info["class"] == 61: - if info["index"] == 0: #error class is 61, index 0 (from ref sheet) + + if info["class"] == 61: #error + if info["index"] == 0: #index 0 means error code is payload(from ref sheet) error_code = msg.data[0] #looks like the message sends error code (error code is mapped to LEDs in ref.) self.error_status.emit(info["dev_id"], error_code) #sends values to update_led_status - self.telemetry_received.emit(info, list(msg.data)) + elif info["class"] == 1: #telem + if info["index"] == 4: #volcurr + vol = msg.data[2] #assume 2 bytes. confirm on tuesday when at sparky. + self.volcurr.emit(info["dev_id"],vol) + def stop(self): self.running = False @@ -73,54 +75,92 @@ def stop(self): class RoverDash(QMainWindow): def __init__(self): super().__init__() - self.setWindowTitle("Sparky - Health Monitor") + self.setWindowTitle("YURS SPARKY DigitalDash") self.resize(1024, 600) self.setStyleSheet("background-color: #0f0f0f; color: white;") central_widget = QWidget() self.setCentralWidget(central_widget) - main_layout = QHBoxLayout(central_widget) + main_layout = QVBoxLayout() + central_widget.setLayout(main_layout) + + # --- MOTOR STATUS --- + self.motor_buttons={} + status_grid = QGridLayout() + + #0 2 3 4 5 are DRIVE axis + status_grid.addWidget(QLabel("Drive Motors"), 0, 0, 1, 6) + + axis0 = QPushButton(f"axis0") + status_grid.addWidget(axis0,1,0) + self.motor_buttons[0] = axis0 - # --- LEFT SIDE: MOTOR STATUS --- - left_column = QVBoxLayout() - self.motor_grid = QGridLayout() - self.motor_buttons = {} - for i in range(12): #create buttons btn1 - 12 - dev_id = i + 1 - btn = QPushButton(f"axis{dev_id}") - btn.setFixedSize(60, 60) - # btn.setStyleSheet(self.get_style("red")) # Default to red until heartbeat - self.motor_grid.addWidget(btn, i // 4, i % 4) - self.motor_buttons[dev_id] = btn + axis2 = QPushButton(f"axis2") + status_grid.addWidget(axis2,1,1) + self.motor_buttons[2] = axis2 + axis3 = QPushButton(f"axis3") + status_grid.addWidget(axis3,1,2) + self.motor_buttons[3] = axis3 - left_column.addLayout(self.motor_grid) - main_layout.addLayout(left_column, 1) + axis4 = QPushButton(f"axis4") + status_grid.addWidget(axis4,1,3) + self.motor_buttons[4] = axis4 - # --- RIGHT SIDE: GRAPHS --- - graph_column = QVBoxLayout() - self.chart_grid = QGridLayout() - self.series_temp = QLineSeries() - self.chart_temp_view = self.create_graph("Temp (°C)", self.series_temp) - self.chart_grid.addWidget(self.chart_temp_view) + axis5 = QPushButton(f"axis5") + status_grid.addWidget(axis5,1,4) + self.motor_buttons[5] = axis5 + + #ARM axis' are 6 7 8 - self.series_current = QLineSeries() #cv represents current - voltage - self.series_current.setName("Current (A)") - self.series_current.setColor(QColor("#00e5ff")) # Cyan-ish for current + status_grid.addWidget(QLabel("ARMs"), 2, 0, 1, 6) - self.series_voltage = QLineSeries() - self.series_voltage.setName("Voltage (V))") - self.series_voltage.setColor(QColor("#dc300e")) # red-ish for voltage + axis6 = QPushButton(f"axis6") + status_grid.addWidget(axis6,3,0) + self.motor_buttons[6] = axis6 - self.chart_currVol_view = self.create_graph("Current & Voltage", [self.series_voltage, self.series_current]) - self.chart_grid.addWidget(self.chart_currVol_view) + axis7 = QPushButton(f"axis7") + status_grid.addWidget(axis7,3,1) + self.motor_buttons[7] = axis7 - graph_column.addLayout(self.chart_grid) - main_layout.addLayout(graph_column, 2) + axis8 = QPushButton(f"axis8") + status_grid.addWidget(axis8,3,2) + self.motor_buttons[8] = axis8 + + #GRIPPER 9 10 11 12 + status_grid.addWidget(QLabel("Grippers"), 4, 0, 1, 6) + + axis9 = QPushButton(f"axis9") + status_grid.addWidget(axis9,5,0) + self.motor_buttons[9] = axis9 + + axis10 = QPushButton(f"axis10") + status_grid.addWidget(axis10,5,1) + self.motor_buttons[10] = axis10 + + axis11 = QPushButton(f"axis11") + status_grid.addWidget(axis11,5,2) + self.motor_buttons[11] = axis11 + + axis12 = QPushButton(f"axis12") + status_grid.addWidget(axis12,5,3) + self.motor_buttons[12] = axis12 + - self.data_count = 0 + #SCIENCE 13 14 + status_grid.addWidget(QLabel("Science"), 6, 0, 1, 6) + + axis13 = QPushButton(f"axis13") + status_grid.addWidget(axis13,7,0) + self.motor_buttons[13] = axis13 + + axis14 = QPushButton(f"axis14") + status_grid.addWidget(axis14,7,1) + self.motor_buttons[14] = axis14 + + main_layout.addLayout(status_grid) # --- WORKER THREAD --- self.worker = CommWorker() @@ -131,16 +171,17 @@ def __init__(self): self.thread.started.connect(self.worker.run) self.worker.error_status.connect(self.update_led_status) - self.worker.telemetry_received.connect(self.process_can_data) + self.worker.volcurr.connect(self.update_volcurr) self.button_states = {} self.blink_timers = {} + self.voltages = {} self.thread.start() # --- LOGIC --- - + def update_led_status(self, dev_id, error_code): color1, color2, speed = LED_MAP.get(error_code, ("red", None, "solid")) @@ -200,58 +241,16 @@ def get_style(self, color): font-weight: bold; }} """ - - - def create_graph(self, title, series_input): - chart = QChart() - chart.setTitle(title) - chart.setTheme(QChart.ChartThemeDark) - - # Ensure we are working with a list even if one series is passed - series_list = series_input if isinstance(series_input, list) else [series_input] - - axis_x = QValueAxis() - axis_x.setRange(0, 100) - chart.addAxis(axis_x, Qt.AlignBottom) - - axis_y = QValueAxis() - axis_y.setRange(0, 100) # Adjust based on your battery/motor limits - chart.addAxis(axis_y, Qt.AlignLeft) - - for s in series_list: - chart.addSeries(s) - s.attachAxis(axis_x) - s.attachAxis(axis_y) - - view = QChartView(chart) - view.setRenderHint(view.renderHints().Antialiasing) - return view - - def process_can_data(self, info, data): - #WORK IN PROGRESS - dev_id = info['dev_id'] - idx = info['index'] - cls = info['class'] - - if cls == 1: - if idx == 3 and dev_id == 6 and len(data) >= 1: # Temperature - val = data[0] - self.series_temp.append(self.data_count, val) - self.data_count += 1 - if idx==4 and dev_id == 6 and len(data)>=1: #current Voltage - vol=data[0] - amp=data[1] - self.series_current.append(self.data_count, vol) - self.series_voltage.append(self.data_count, amp) - - #Auto-scroll & Increment X-Axis - self.data_count += 1 - if self.data_count > 100: - new_min, new_max = self.data_count - 100, self.data_count - self.chart_temp_view.chart().axisX().setRange(new_min, new_max) - self.chart_currVol_view.chart().axisX().setRange(new_min, new_max) + def update_volcurr(self, dev_id, vol): + btn = self.motor_buttons.get(dev_id) + self.voltages[dev_id] = vol + if btn is None: + return + + btn.setText(f"axis{dev_id}\n{vol}V") + def closeEvent(self, event): self.worker.stop() # tell worker loop to exit self.thread.quit() # stop event loop From 508b62f0e8abd8e38eb6c7eaea70f50f20892643 Mon Sep 17 00:00:00 2001 From: IdkHowToCode123 Date: Tue, 23 Jun 2026 20:01:56 -0400 Subject: [PATCH 5/8] Added Little-Endian Concatenation --- main.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/main.py b/main.py index a26ffdf..6af7e25 100644 --- a/main.py +++ b/main.py @@ -53,7 +53,9 @@ def run(self): self.error_status.emit(info["dev_id"], error_code) #sends values to update_led_status elif info["class"] == 1: #telem if info["index"] == 4: #volcurr - vol = msg.data[2] #assume 2 bytes. confirm on tuesday when at sparky. + volp1 = msg.data[0] + volp2 = msg.data[1] + vol = (volp2<<8) | (volp1) #MSB Little-Endian Concatenation self.volcurr.emit(info["dev_id"],vol) def stop(self): From 7791e9c788e78ba4884871ca267ff7d0800bc200 Mon Sep 17 00:00:00 2001 From: unknown Date: Fri, 10 Jul 2026 22:48:58 -0400 Subject: [PATCH 6/8] Cleaned up & Future-Proofed :) --- .vscode/settings.json | 3 + CANTesterUnit.py | 68 ++++++++++ Mac-CANTest.py | 49 ------- main.py | 294 ++++++++++++++++++++---------------------- 4 files changed, 209 insertions(+), 205 deletions(-) create mode 100644 .vscode/settings.json create mode 100644 CANTesterUnit.py delete mode 100644 Mac-CANTest.py diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..c9ebf2d --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "python-envs.defaultEnvManager": "ms-python.python:system" +} \ No newline at end of file diff --git a/CANTesterUnit.py b/CANTesterUnit.py new file mode 100644 index 0000000..bf87eff --- /dev/null +++ b/CANTesterUnit.py @@ -0,0 +1,68 @@ +import time +import can + + +def build_can_id(dev_id, msg_class, index, manufacturer=8, device_type=2): + return ( + (device_type << 24) + | (manufacturer << 16) + | (msg_class << 10) + | (index << 6) + | dev_id + ) + + +bus = can.interface.Bus( + interface="slcan", + channel="rover-test", + bitrate=1_000_000, +) + +try: + while True: + # Send voltage telemetry to every configured motor + for dev_id in [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14]: + voltage = 1200 + dev_id + + voltage_msg = can.Message( + arbitration_id=build_can_id( + dev_id=dev_id, + msg_class=1, + index=4, + ), + data=[ + voltage & 0xFF, + (voltage >> 8) & 0xFF, + 0, + 0, + 0, + 0, + 0, + 0, + ], + is_extended_id=True, + ) + + bus.send(voltage_msg) + + # Send a different LED/error state to axis 6 + error_msg = can.Message( + arbitration_id=build_can_id( + dev_id=6, + msg_class=61, + index=0, + ), + data=[3, 0, 0, 0, 0, 0, 0, 0], + is_extended_id=True, + ) + + bus.send(error_msg) + + print("Sent test telemetry and error frames") + time.sleep(1) + +except KeyboardInterrupt: + print("Stopping CAN test sender") + +finally: + bus.shutdown() diff --git a/Mac-CANTest.py b/Mac-CANTest.py deleted file mode 100644 index c39839e..0000000 --- a/Mac-CANTest.py +++ /dev/null @@ -1,49 +0,0 @@ -import can -import time -import random - -def generate_fake_can(): - # Use 'virtual' for macOS compatibility - try: - bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast') - except: - # If vcan0 isn't defined in your configs, this usually just works - bus = can.interface.Bus(channel='224.0.0.1', interface='udp_multicast') - - print("Bus started. Sending fake Rover telemetry...") - - # Constants from your whiteboard - TYPE = 2 - MANU = 8 - DEV_ID = 6 # Targeting Motor 6 as per your UI code - - count = 0 - while True: - # 1. Generate Heartbeat (Class 63, Index 0) - # ID Construction: (Type << 24) | (Manu << 16) | (Class << 10) | (Index << 6) | DevID - heartbeat_id = (TYPE << 24) | (MANU << 16) | (63 << 10) | (0 << 6) | DEV_ID - msg_hb = can.Message(arbitration_id=heartbeat_id, data=[1], is_extended_id=True) - bus.send(msg_hb) - #print(msg_hb) - - #another heartbeat - heartbeat_id = (TYPE << 24) | (MANU << 16) | (63 << 10) | (0 << 6) | 9 - msg_hb = can.Message(arbitration_id=heartbeat_id, data=[1], is_extended_id=True) - bus.send(msg_hb) - # 2. Generate Temp (Class 1, Index 3) - temp_id = (TYPE << 24) | (MANU << 16) | (1 << 10) | (3 << 6) | DEV_ID - fake_temp = int(40 + 10 * random.random()) # 40-50 degrees - msg_temp = can.Message(arbitration_id=temp_id, data=[fake_temp], is_extended_id=True) - bus.send(msg_temp) - - # 3. Generate Volt/Curr (Class 1, Index 4) - power_id = (TYPE << 24) | (MANU << 16) | (1 << 10) | (4 << 6) | DEV_ID - fake_volt = int(20 + random.random() * 4) # 20-24V - fake_curr = int(random.random() * 50) # 0-50A - msg_power = can.Message(arbitration_id=power_id, data=[fake_volt, fake_curr], is_extended_id=True) - bus.send(msg_power) - - time.sleep(0.1) # Send at 10Hz - -if __name__ == "__main__": - generate_fake_can() \ No newline at end of file diff --git a/main.py b/main.py index 6af7e25..dfb452d 100644 --- a/main.py +++ b/main.py @@ -1,13 +1,13 @@ import sys import can -from PySide6.QtWidgets import (QApplication, QLabel, QMainWindow, QWidget, QVBoxLayout, +import time +from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QPushButton, QGridLayout) from PySide6.QtCore import QThread, Signal, QObject, QTimer #this commworker handles all of the CAN communication. class CommWorker(QObject): - error_status = Signal(int, int) - volcurr = Signal(int,int) + package = Signal(int, int,str) def run(self): try: @@ -50,18 +50,63 @@ def run(self): if info["class"] == 61: #error if info["index"] == 0: #index 0 means error code is payload(from ref sheet) error_code = msg.data[0] #looks like the message sends error code (error code is mapped to LEDs in ref.) - self.error_status.emit(info["dev_id"], error_code) #sends values to update_led_status + self.package.emit(info["dev_id"], error_code,"Error") #sends values to update_led_status elif info["class"] == 1: #telem if info["index"] == 4: #volcurr volp1 = msg.data[0] volp2 = msg.data[1] vol = (volp2<<8) | (volp1) #MSB Little-Endian Concatenation - self.volcurr.emit(info["dev_id"],vol) + self.package.emit(info["dev_id"],vol,"Voltage") def stop(self): self.running = False - +#this FakeWorker is a testing unit. see CANTesterUnit.py +class FakeWorker(QObject): + package = Signal(int, int, str) + + def run(self): + self.timer = QTimer() + self.timer.timeout.connect(self.send) + self.timer.start(1000) + + self.count = 0 + + def send(self): + self.count += 1 + self.package.emit(1, 1220 + self.count, "Voltage") + self.package.emit(7, 1200 + self.count, "Voltage") + self.package.emit(6, 1180 + self.count, "Voltage") + self.package.emit(9, 1220 + self.count, "Voltage") + error = (self.count % 9) + 1 + self.package.emit(6, 7, "Error") + self.package.emit(8, 7, "Error") + self.package.emit(11, 7, "Error") + self.package.emit(4, 7, "Error") + self.package.emit(2, 2, "Error") #tested :) + def stop(self): + self.timer.stop() +#which button is where? +button_positions = { + 0: (1, 0), + 2: (1, 1), + 3: (1, 2), + 4: (1, 3), + 5: (1, 4), + + 6: (3, 0), + 7: (3, 1), + 8: (3, 2), + + 9: (5, 0), + 10: (5, 1), + 11: (5, 2), + 12: (5, 3), + + 13: (7, 0), + 14: (7, 1), +} +#Cartier LED_MAP = { 1: ("red", "blue", "slow"), 2: ("red", "cyan", "slow"), @@ -74,10 +119,11 @@ def stop(self): 9: ("green", "green", "normal"), } + class RoverDash(QMainWindow): def __init__(self): super().__init__() - self.setWindowTitle("YURS SPARKY DigitalDash") + self.setWindowTitle("YURS SPARKY 2026 telemoled") self.resize(1024, 600) self.setStyleSheet("background-color: #0f0f0f; color: white;") @@ -87,93 +133,25 @@ def __init__(self): main_layout = QVBoxLayout() central_widget.setLayout(main_layout) - # --- MOTOR STATUS --- - self.motor_buttons={} - status_grid = QGridLayout() - - #0 2 3 4 5 are DRIVE axis - status_grid.addWidget(QLabel("Drive Motors"), 0, 0, 1, 6) - - axis0 = QPushButton(f"axis0") - status_grid.addWidget(axis0,1,0) - self.motor_buttons[0] = axis0 - - - axis2 = QPushButton(f"axis2") - status_grid.addWidget(axis2,1,1) - self.motor_buttons[2] = axis2 - - axis3 = QPushButton(f"axis3") - status_grid.addWidget(axis3,1,2) - self.motor_buttons[3] = axis3 - - axis4 = QPushButton(f"axis4") - status_grid.addWidget(axis4,1,3) - self.motor_buttons[4] = axis4 - - axis5 = QPushButton(f"axis5") - status_grid.addWidget(axis5,1,4) - self.motor_buttons[5] = axis5 - - #ARM axis' are 6 7 8 - - status_grid.addWidget(QLabel("ARMs"), 2, 0, 1, 6) - - axis6 = QPushButton(f"axis6") - status_grid.addWidget(axis6,3,0) - self.motor_buttons[6] = axis6 - - axis7 = QPushButton(f"axis7") - status_grid.addWidget(axis7,3,1) - self.motor_buttons[7] = axis7 - - axis8 = QPushButton(f"axis8") - status_grid.addWidget(axis8,3,2) - self.motor_buttons[8] = axis8 - - #GRIPPER 9 10 11 12 - status_grid.addWidget(QLabel("Grippers"), 4, 0, 1, 6) - - axis9 = QPushButton(f"axis9") - status_grid.addWidget(axis9,5,0) - self.motor_buttons[9] = axis9 - - axis10 = QPushButton(f"axis10") - status_grid.addWidget(axis10,5,1) - self.motor_buttons[10] = axis10 - - axis11 = QPushButton(f"axis11") - status_grid.addWidget(axis11,5,2) - self.motor_buttons[11] = axis11 - - axis12 = QPushButton(f"axis12") - status_grid.addWidget(axis12,5,3) - self.motor_buttons[12] = axis12 - + self.status_grid = QGridLayout() + self.motor_buttons = {} + # --- MOTOR DEFs --- + for dev_id, position in button_positions.items(): + btn = QPushButton(f"axis{dev_id}") + self.status_grid.addWidget(btn, *position) + self.motor_buttons[dev_id] = btn - #SCIENCE 13 14 - status_grid.addWidget(QLabel("Science"), 6, 0, 1, 6) - - axis13 = QPushButton(f"axis13") - status_grid.addWidget(axis13,7,0) - self.motor_buttons[13] = axis13 - - axis14 = QPushButton(f"axis14") - status_grid.addWidget(axis14,7,1) - self.motor_buttons[14] = axis14 - - main_layout.addLayout(status_grid) + main_layout.addLayout(self.status_grid) # --- WORKER THREAD --- - self.worker = CommWorker() + self.worker = FakeWorker() self.thread = QThread() self.worker.moveToThread(self.thread) self.thread.started.connect(self.worker.run) - self.worker.error_status.connect(self.update_led_status) - self.worker.volcurr.connect(self.update_volcurr) + self.worker.package.connect(self.package_received) self.button_states = {} self.blink_timers = {} @@ -182,83 +160,87 @@ def __init__(self): self.thread.start() # --- LOGIC --- - - - def update_led_status(self, dev_id, error_code): - color1, color2, speed = LED_MAP.get(error_code, ("red", None, "solid")) - - btn = self.motor_buttons.get(dev_id) - if btn is None: - return - - if dev_id in self.blink_timers: - self.blink_timers[dev_id].stop() - del self.blink_timers[dev_id] - - if speed == "solid" or color2 is None: - btn.setStyleSheet(self.get_style(color1)) - return - - intervals = { - "normal": 250, - "slow": 500, + def get_style(self, color): + colors = { + "red": "#ff3333", + "green": "#00ff88", + "blue": "#3399ff", + "cyan": "#00e5ff", + "magenta": "#ff33cc", + "yellow": "#ffff33", + "off": "#222222", } + hex_color = colors.get(color, "#ff3333") + return f""" + QPushButton {{ + background-color: {hex_color}; + color: black; + border-radius: 8px; + font-weight: bold; + }} """ + + def package_received(self, dev_id, value, type): + if (type == "Error"): + color1, color2, speed = LED_MAP.get(value, ("red", None, "solid")) + + btn = self.motor_buttons.get(dev_id) + if btn is None: + return + + if dev_id in self.blink_timers: + self.blink_timers[dev_id].stop() + del self.blink_timers[dev_id] + + if speed == "solid" or color2 is None: + btn.setStyleSheet(self.get_style(color1)) + return + + intervals = { + "normal": 250, + "slow": 500, + } + + interval = intervals.get(speed, 500) + + state = 1 + + def blink(): + nonlocal state + state = not state + + color = color1 if state else color2 + btn.setStyleSheet(self.get_style(color)) + + timer = QTimer(self) + timer.timeout.connect(blink) + timer.start(interval) + + self.blink_timers[dev_id] = timer + blink() - interval = intervals.get(speed, 500) - - state = 1 - - def blink(): - nonlocal state - state = not state - - color = color1 if state else color2 - btn.setStyleSheet(self.get_style(color)) - - timer = QTimer(self) - timer.timeout.connect(blink) - timer.start(interval) - - self.blink_timers[dev_id] = timer - blink() - - def get_style(self, color): - colors = { - "red": "#ff3333", - "green": "#00ff88", - "blue": "#3399ff", - "cyan": "#00e5ff", - "magenta": "#ff33cc", - "yellow": "#ffff33", - "off": "#222222", - } - - hex_color = colors.get(color, "#ff3333") - - return f""" - QPushButton {{ - background-color: {hex_color}; - color: black; - border-radius: 8px; - font-weight: bold; - }} - """ - def update_volcurr(self, dev_id, vol): - btn = self.motor_buttons.get(dev_id) - self.voltages[dev_id] = vol - if btn is None: - return - - btn.setText(f"axis{dev_id}\n{vol}V") + + if (type == "Voltage"): + btn = self.motor_buttons.get(dev_id) + self.voltages[dev_id] = value + if btn is None: + return + if dev_id <= 5: + btn.setText(f"axis{dev_id} DRIVE MOTOR\n{value} V") + if 6 <= dev_id <= 8: + btn.setText(f"axis{dev_id} ARMS! \n{value} V") + if 9 <= dev_id <= 12: + btn.setText(f"axis{dev_id} GRIPPERS ;) \n{value} V") + if 13 <= dev_id <= 14: + btn.setText(f"axis{dev_id}--- SCIENCE ---! \n{value} V") - + def closeEvent(self, event): - self.worker.stop() # tell worker loop to exit - self.thread.quit() # stop event loop - self.thread.wait() # wait for thread to finish + self.worker.stop() # tell worker loop to exit + self.thread.quit() # stop event loop + self.thread.wait() # wait for thread to finish - event.accept() + event.accept() if __name__ == "__main__": app = QApplication(sys.argv) From 2f1fad2f8e6ba46b3b2a2c4d2e7ee53c3d222f30 Mon Sep 17 00:00:00 2001 From: Leonardo314159 Date: Fri, 10 Jul 2026 22:53:51 -0400 Subject: [PATCH 7/8] Cleaned up. Included All Testers. Future Proofed :) --- .vscode/settings.json | 4 +- CANTesterUnit.py | 136 ++++++------ main.py | 496 +++++++++++++++++++++--------------------- 3 files changed, 318 insertions(+), 318 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index c9ebf2d..d84f211 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,3 +1,3 @@ -{ - "python-envs.defaultEnvManager": "ms-python.python:system" +{ + "python-envs.defaultEnvManager": "ms-python.python:system" } \ No newline at end of file diff --git a/CANTesterUnit.py b/CANTesterUnit.py index bf87eff..ea5deb5 100644 --- a/CANTesterUnit.py +++ b/CANTesterUnit.py @@ -1,68 +1,68 @@ -import time -import can - - -def build_can_id(dev_id, msg_class, index, manufacturer=8, device_type=2): - return ( - (device_type << 24) - | (manufacturer << 16) - | (msg_class << 10) - | (index << 6) - | dev_id - ) - - -bus = can.interface.Bus( - interface="slcan", - channel="rover-test", - bitrate=1_000_000, -) - -try: - while True: - # Send voltage telemetry to every configured motor - for dev_id in [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14]: - voltage = 1200 + dev_id - - voltage_msg = can.Message( - arbitration_id=build_can_id( - dev_id=dev_id, - msg_class=1, - index=4, - ), - data=[ - voltage & 0xFF, - (voltage >> 8) & 0xFF, - 0, - 0, - 0, - 0, - 0, - 0, - ], - is_extended_id=True, - ) - - bus.send(voltage_msg) - - # Send a different LED/error state to axis 6 - error_msg = can.Message( - arbitration_id=build_can_id( - dev_id=6, - msg_class=61, - index=0, - ), - data=[3, 0, 0, 0, 0, 0, 0, 0], - is_extended_id=True, - ) - - bus.send(error_msg) - - print("Sent test telemetry and error frames") - time.sleep(1) - -except KeyboardInterrupt: - print("Stopping CAN test sender") - -finally: - bus.shutdown() +import time +import can + + +def build_can_id(dev_id, msg_class, index, manufacturer=8, device_type=2): + return ( + (device_type << 24) + | (manufacturer << 16) + | (msg_class << 10) + | (index << 6) + | dev_id + ) + + +bus = can.interface.Bus( + interface="slcan", + channel="rover-test", + bitrate=1_000_000, +) + +try: + while True: + # Send voltage telemetry to every configured motor + for dev_id in [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14]: + voltage = 1200 + dev_id + + voltage_msg = can.Message( + arbitration_id=build_can_id( + dev_id=dev_id, + msg_class=1, + index=4, + ), + data=[ + voltage & 0xFF, + (voltage >> 8) & 0xFF, + 0, + 0, + 0, + 0, + 0, + 0, + ], + is_extended_id=True, + ) + + bus.send(voltage_msg) + + # Send a different LED/error state to axis 6 + error_msg = can.Message( + arbitration_id=build_can_id( + dev_id=6, + msg_class=61, + index=0, + ), + data=[3, 0, 0, 0, 0, 0, 0, 0], + is_extended_id=True, + ) + + bus.send(error_msg) + + print("Sent test telemetry and error frames") + time.sleep(1) + +except KeyboardInterrupt: + print("Stopping CAN test sender") + +finally: + bus.shutdown() diff --git a/main.py b/main.py index dfb452d..defaa5f 100644 --- a/main.py +++ b/main.py @@ -1,249 +1,249 @@ -import sys -import can -import time -from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, - QPushButton, QGridLayout) -from PySide6.QtCore import QThread, Signal, QObject, QTimer - -#this commworker handles all of the CAN communication. -class CommWorker(QObject): - package = Signal(int, int,str) - - def run(self): - try: - #Masks out only communication from SPARK devices. - target_id = (2 << 24) | (8 << 16) - mask = 0x1FFF0000 - - filters = [{ - "can_id": target_id, - "can_mask": mask, - "extended": True - }] - - bus = can.interface.Bus( - interface='slcan', - channel='rover-test', - bitrate=1000000, - filters=filters - ) - except Exception as e: - print(f"CAN Bus Error: {e}") - return - self.running = True - - while self.running: - msg = bus.recv(timeout=0.01) - print(msg) - if msg is None: - continue - - can_id = msg.arbitration_id - info = { - "dev_id": (can_id >> 0) & 0x3F, #device ID of sparkmax [gets passed to btn] - "index": (can_id >> 6) & 0x0F, - "class": (can_id >> 10) & 0x3F, - "manu": (can_id >> 16) & 0xFF, - "type": (can_id >> 24) & 0x1F, - } - - if info["class"] == 61: #error - if info["index"] == 0: #index 0 means error code is payload(from ref sheet) - error_code = msg.data[0] #looks like the message sends error code (error code is mapped to LEDs in ref.) - self.package.emit(info["dev_id"], error_code,"Error") #sends values to update_led_status - elif info["class"] == 1: #telem - if info["index"] == 4: #volcurr - volp1 = msg.data[0] - volp2 = msg.data[1] - vol = (volp2<<8) | (volp1) #MSB Little-Endian Concatenation - self.package.emit(info["dev_id"],vol,"Voltage") - - def stop(self): - self.running = False -#this FakeWorker is a testing unit. see CANTesterUnit.py -class FakeWorker(QObject): - package = Signal(int, int, str) - - def run(self): - self.timer = QTimer() - self.timer.timeout.connect(self.send) - self.timer.start(1000) - - self.count = 0 - - def send(self): - self.count += 1 - self.package.emit(1, 1220 + self.count, "Voltage") - self.package.emit(7, 1200 + self.count, "Voltage") - self.package.emit(6, 1180 + self.count, "Voltage") - self.package.emit(9, 1220 + self.count, "Voltage") - error = (self.count % 9) + 1 - self.package.emit(6, 7, "Error") - self.package.emit(8, 7, "Error") - self.package.emit(11, 7, "Error") - self.package.emit(4, 7, "Error") - self.package.emit(2, 2, "Error") #tested :) - - def stop(self): - self.timer.stop() -#which button is where? -button_positions = { - 0: (1, 0), - 2: (1, 1), - 3: (1, 2), - 4: (1, 3), - 5: (1, 4), - - 6: (3, 0), - 7: (3, 1), - 8: (3, 2), - - 9: (5, 0), - 10: (5, 1), - 11: (5, 2), - 12: (5, 3), - - 13: (7, 0), - 14: (7, 1), -} -#Cartier -LED_MAP = { - 1: ("red", "blue", "slow"), - 2: ("red", "cyan", "slow"), - 3: ("red", "green", "slow"), - 4: ("red", "magenta", "slow"), - 5: ("red", "yellow", "slow"), - 6: ("cyan", "cyan", "normal"), - 7: ("cyan", "cyan", "solid"), - 8: ("red", "cyan", "normal"), - 9: ("green", "green", "normal"), -} - - -class RoverDash(QMainWindow): - def __init__(self): - super().__init__() - self.setWindowTitle("YURS SPARKY 2026 telemoled") - self.resize(1024, 600) - self.setStyleSheet("background-color: #0f0f0f; color: white;") - - central_widget = QWidget() - self.setCentralWidget(central_widget) - - main_layout = QVBoxLayout() - central_widget.setLayout(main_layout) - - self.status_grid = QGridLayout() - self.motor_buttons = {} - # --- MOTOR DEFs --- - for dev_id, position in button_positions.items(): - btn = QPushButton(f"axis{dev_id}") - self.status_grid.addWidget(btn, *position) - self.motor_buttons[dev_id] = btn - - main_layout.addLayout(self.status_grid) - - # --- WORKER THREAD --- - self.worker = FakeWorker() - self.thread = QThread() - - self.worker.moveToThread(self.thread) - - self.thread.started.connect(self.worker.run) - - self.worker.package.connect(self.package_received) - - self.button_states = {} - self.blink_timers = {} - self.voltages = {} - - self.thread.start() - - # --- LOGIC --- - def get_style(self, color): - colors = { - "red": "#ff3333", - "green": "#00ff88", - "blue": "#3399ff", - "cyan": "#00e5ff", - "magenta": "#ff33cc", - "yellow": "#ffff33", - "off": "#222222", - } - hex_color = colors.get(color, "#ff3333") - return f""" - QPushButton {{ - background-color: {hex_color}; - color: black; - border-radius: 8px; - font-weight: bold; - }} """ - - def package_received(self, dev_id, value, type): - if (type == "Error"): - color1, color2, speed = LED_MAP.get(value, ("red", None, "solid")) - - btn = self.motor_buttons.get(dev_id) - if btn is None: - return - - if dev_id in self.blink_timers: - self.blink_timers[dev_id].stop() - del self.blink_timers[dev_id] - - if speed == "solid" or color2 is None: - btn.setStyleSheet(self.get_style(color1)) - return - - intervals = { - "normal": 250, - "slow": 500, - } - - interval = intervals.get(speed, 500) - - state = 1 - - def blink(): - nonlocal state - state = not state - - color = color1 if state else color2 - btn.setStyleSheet(self.get_style(color)) - - timer = QTimer(self) - timer.timeout.connect(blink) - timer.start(interval) - - self.blink_timers[dev_id] = timer - blink() - - - - if (type == "Voltage"): - btn = self.motor_buttons.get(dev_id) - self.voltages[dev_id] = value - if btn is None: - return - if dev_id <= 5: - btn.setText(f"axis{dev_id} DRIVE MOTOR\n{value} V") - if 6 <= dev_id <= 8: - btn.setText(f"axis{dev_id} ARMS! \n{value} V") - if 9 <= dev_id <= 12: - btn.setText(f"axis{dev_id} GRIPPERS ;) \n{value} V") - if 13 <= dev_id <= 14: - btn.setText(f"axis{dev_id}--- SCIENCE ---! \n{value} V") - - - def closeEvent(self, event): - self.worker.stop() # tell worker loop to exit - self.thread.quit() # stop event loop - self.thread.wait() # wait for thread to finish - - event.accept() - -if __name__ == "__main__": - app = QApplication(sys.argv) - window = RoverDash() - window.show() +import sys +import can +import time +from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, + QPushButton, QGridLayout) +from PySide6.QtCore import QThread, Signal, QObject, QTimer + +#this commworker handles all of the CAN communication. +class CommWorker(QObject): + package = Signal(int, int,str) + + def run(self): + try: + #Masks out only communication from SPARK devices. + target_id = (2 << 24) | (8 << 16) + mask = 0x1FFF0000 + + filters = [{ + "can_id": target_id, + "can_mask": mask, + "extended": True + }] + + bus = can.interface.Bus( + interface='slcan', + channel='rover-test', + bitrate=1000000, + filters=filters + ) + except Exception as e: + print(f"CAN Bus Error: {e}") + return + self.running = True + + while self.running: + msg = bus.recv(timeout=0.01) + print(msg) + if msg is None: + continue + + can_id = msg.arbitration_id + info = { + "dev_id": (can_id >> 0) & 0x3F, #device ID of sparkmax [gets passed to btn] + "index": (can_id >> 6) & 0x0F, + "class": (can_id >> 10) & 0x3F, + "manu": (can_id >> 16) & 0xFF, + "type": (can_id >> 24) & 0x1F, + } + + if info["class"] == 61: #error + if info["index"] == 0: #index 0 means error code is payload(from ref sheet) + error_code = msg.data[0] #looks like the message sends error code (error code is mapped to LEDs in ref.) + self.package.emit(info["dev_id"], error_code,"Error") #sends values to update_led_status + elif info["class"] == 1: #telem + if info["index"] == 4: #volcurr + volp1 = msg.data[0] + volp2 = msg.data[1] + vol = (volp2<<8) | (volp1) #MSB Little-Endian Concatenation + self.package.emit(info["dev_id"],vol,"Voltage") + + def stop(self): + self.running = False +#this FakeWorker is a testing unit. see CANTesterUnit.py +class FakeWorker(QObject): + package = Signal(int, int, str) + + def run(self): + self.timer = QTimer() + self.timer.timeout.connect(self.send) + self.timer.start(1000) + + self.count = 0 + + def send(self): + self.count += 1 + self.package.emit(1, 1220 + self.count, "Voltage") + self.package.emit(7, 1200 + self.count, "Voltage") + self.package.emit(6, 1180 + self.count, "Voltage") + self.package.emit(9, 1220 + self.count, "Voltage") + error = (self.count % 9) + 1 + self.package.emit(6, 7, "Error") + self.package.emit(8, 7, "Error") + self.package.emit(11, 7, "Error") + self.package.emit(4, 7, "Error") + self.package.emit(2, 2, "Error") #tested :) + + def stop(self): + self.timer.stop() +#which button is where? +button_positions = { + 0: (1, 0), + 2: (1, 1), + 3: (1, 2), + 4: (1, 3), + 5: (1, 4), + + 6: (3, 0), + 7: (3, 1), + 8: (3, 2), + + 9: (5, 0), + 10: (5, 1), + 11: (5, 2), + 12: (5, 3), + + 13: (7, 0), + 14: (7, 1), +} +#Cartier +LED_MAP = { + 1: ("red", "blue", "slow"), + 2: ("red", "cyan", "slow"), + 3: ("red", "green", "slow"), + 4: ("red", "magenta", "slow"), + 5: ("red", "yellow", "slow"), + 6: ("cyan", "cyan", "normal"), + 7: ("cyan", "cyan", "solid"), + 8: ("red", "cyan", "normal"), + 9: ("green", "green", "normal"), +} + + +class RoverDash(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("YURS SPARKY 2026 telemoled") + self.resize(1024, 600) + self.setStyleSheet("background-color: #0f0f0f; color: white;") + + central_widget = QWidget() + self.setCentralWidget(central_widget) + + main_layout = QVBoxLayout() + central_widget.setLayout(main_layout) + + self.status_grid = QGridLayout() + self.motor_buttons = {} + # --- MOTOR DEFs --- + for dev_id, position in button_positions.items(): + btn = QPushButton(f"axis{dev_id}") + self.status_grid.addWidget(btn, *position) + self.motor_buttons[dev_id] = btn + + main_layout.addLayout(self.status_grid) + + # --- WORKER THREAD --- + self.worker = FakeWorker() + self.thread = QThread() + + self.worker.moveToThread(self.thread) + + self.thread.started.connect(self.worker.run) + + self.worker.package.connect(self.package_received) + + self.button_states = {} + self.blink_timers = {} + self.voltages = {} + + self.thread.start() + + # --- LOGIC --- + def get_style(self, color): + colors = { + "red": "#ff3333", + "green": "#00ff88", + "blue": "#3399ff", + "cyan": "#00e5ff", + "magenta": "#ff33cc", + "yellow": "#ffff33", + "off": "#222222", + } + hex_color = colors.get(color, "#ff3333") + return f""" + QPushButton {{ + background-color: {hex_color}; + color: black; + border-radius: 8px; + font-weight: bold; + }} """ + + def package_received(self, dev_id, value, type): + if (type == "Error"): + color1, color2, speed = LED_MAP.get(value, ("red", None, "solid")) + + btn = self.motor_buttons.get(dev_id) + if btn is None: + return + + if dev_id in self.blink_timers: + self.blink_timers[dev_id].stop() + del self.blink_timers[dev_id] + + if speed == "solid" or color2 is None: + btn.setStyleSheet(self.get_style(color1)) + return + + intervals = { + "normal": 250, + "slow": 500, + } + + interval = intervals.get(speed, 500) + + state = 1 + + def blink(): + nonlocal state + state = not state + + color = color1 if state else color2 + btn.setStyleSheet(self.get_style(color)) + + timer = QTimer(self) + timer.timeout.connect(blink) + timer.start(interval) + + self.blink_timers[dev_id] = timer + blink() + + + + if (type == "Voltage"): + btn = self.motor_buttons.get(dev_id) + self.voltages[dev_id] = value + if btn is None: + return + if dev_id <= 5: + btn.setText(f"axis{dev_id} DRIVE MOTOR\n{value} V") + if 6 <= dev_id <= 8: + btn.setText(f"axis{dev_id} ARMS! \n{value} V") + if 9 <= dev_id <= 12: + btn.setText(f"axis{dev_id} GRIPPERS ;) \n{value} V") + if 13 <= dev_id <= 14: + btn.setText(f"axis{dev_id}--- SCIENCE ---! \n{value} V") + + + def closeEvent(self, event): + self.worker.stop() # tell worker loop to exit + self.thread.quit() # stop event loop + self.thread.wait() # wait for thread to finish + + event.accept() + +if __name__ == "__main__": + app = QApplication(sys.argv) + window = RoverDash() + window.show() sys.exit(app.exec()) \ No newline at end of file From ede080fb4fbe61c437a854e7c1460de0a69758bc Mon Sep 17 00:00:00 2001 From: Leonardo Peragine Date: Fri, 10 Jul 2026 23:05:03 -0400 Subject: [PATCH 8/8] Forgot time.py --- main.py | 1 - 1 file changed, 1 deletion(-) diff --git a/main.py b/main.py index defaa5f..1d40a73 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,5 @@ import sys import can -import time from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QPushButton, QGridLayout) from PySide6.QtCore import QThread, Signal, QObject, QTimer