It looks like you're working on a PySide6 application with pyqtgraph for real-time data visualization. Based on your code snippet, you have a main window with graphs that are updated by a QThread through signals. Let me help you structure this properly.
import datetime
from PySide6.QtWidgets import QMainWindow, QWidget, QCheckBox
from PySide6.QtCore import Signal, QThread
import pyqtgraph as pg
class TimeAxisItem(pg.AxisItem):
"""Converts time stamps to pyqtgraph Axis"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def tickStrings(self, values, scale, spacing):
"""converts timestamp to pyqtgraph axis"""
return [
datetime.datetime.fromtimestamp(value).strftime("%Y-%m-%d %H:%M:%S")
for value in values
]
class WorkerThread(QThread):
"""Thread that generates or fetches data"""
data_ready = Signal(object) # Signal to emit new data
def run(self):
while True:
# Simulate data acquisition
data = self.get_new_data()
self.data_ready.emit(data)
self.msleep(1000) # Update every second
def get_new_data(self):
"""Replace this with your actual data acquisition logic"""
import time
import random
return {
'timestamp': time.time(),
'values': [random.random() for _ in range(30)],
'status': random.randint(0, 1)
}
class MainWindow(QMainWindow):
"""Main Window of the Dashboard."""
def __init__(self):
super().__init__()
# Initialize UI elements
self.setup_ui()
self.setup_graphs()
self.setup_thread()
# Initialize checkboxes
self._vfm_tags_checkboxes = []
for i in range(1, 31):
inner_body = self.findChild(QWidget, f"VFM_Tag_No_{i}_body")
if inner_body: # Make sure the widget exists
for checkbox in inner_body.findChildren(QCheckBox):
self._vfm_tags_checkboxes.append(checkbox)
def setup_ui(self):
"""Setup your main window UI here"""
self.setWindowTitle("Real-time Dashboard")
self.resize(1200, 800)
# Create central widget and layout
central_widget = QWidget()
self.setCentralWidget(central_widget)
# You would add your actual UI elements here
# ...
def setup_graphs(self):
"""Initialize the graphs"""
# Graph 1
self.graph1 = pg.PlotWidget(axisItems={'bottom': TimeAxisItem(orientation='bottom')})
self.graph1.setLabel('left', "Value 1")
self.graph1.setLabel('bottom', "Time")
self.graph1.showGrid(x=True, y=True)
self.graph1_curve = self.graph1.plot(pen='y')
# Graph 2
self.graph2 = pg.PlotWidget(axisItems={'bottom': TimeAxisItem(orientation='bottom')})
self.graph2.setLabel('left', "Value 2")
self.graph2.setLabel('bottom', "Time")
self.graph2.showGrid(x=True, y=True)
self.graph2_curve = self.graph2.plot(pen='r')
# Add graphs to your layout (adjust as needed)
layout = self.centralWidget().layout()
layout.addWidget(self.graph1)
layout.addWidget(self.graph2)
def setup_thread(self):
"""Setup and start the worker thread"""
self.worker_thread = WorkerThread()
self.worker_thread.data_ready.connect(self.update_graphs)
self.worker_thread.start()
def update_graphs(self, data):
"""Update graphs with new data"""
# Process data
timestamp = data['timestamp']
values = data['values']
status = data['status']
# Update UI elements based on status
self.update_status_indicators(status)
# Update graph data
# You'll need to maintain your own data buffers
if not hasattr(self, 'time_data'):
self.time_data = []
self.graph1_data = []
self.graph2_data = []
self.time_data.append(timestamp)
self.graph1_data.append(values[0]) # First value for graph1
self.graph2_data.append(values[1]) # Second value for graph2
# Keep only the last N points
max_points = 100
self.time_data = self.time_data[-max_points:]
self.graph1_data = self.graph1_data[-max_points:]
self.graph2_data = self.graph2_data[-max_points:]
# Update plots
self.graph1_curve.setData(self.time_data, self.graph1_data)
self.graph2_curve.setData(self.time_data, self.graph2_data)
def update_status_indicators(self, status):
"""Update checkboxes or other UI elements based on status"""
for checkbox in self._vfm_tags_checkboxes:
# Example: Update checkboxes based on status
checkbox.setChecked(status)
def closeEvent(self, event):
"""Ensure thread is stopped when window is closed"""
self.worker_thread.quit()
self.worker_thread.wait()
super().closeEvent(event)
if __name__ == "__main__":
from PySide6.QtWidgets import QApplication
app = QApplication([])
window = MainWindow()
window.show()
app.exec()
Key points to consider:
- Thread Safety: All Ui updates must happen in the main thread. Your worker thread should only emit signals with data, not perform any Ui operations directly.
- Data Buffering: Maintain buffers for your time series data to limit the number of points displayed.
- Performance: For high-frequency updates, consider:
- UsingÂ
pg.setConfigOptions(useOpenGL=True)
 for hardware acceleration - Limiting the number of points displayed
- UsingÂ
pg.QtGui.QApplication.processEvents()
 sparingly - Error Handling: Add proper error handling in your worker thread to prevent crashes.
- Clean Shutdown: Always properly stop your threads when closing the application.