mirror of
https://github.com/Athemis/PyDSF.git
synced 2025-04-05 14:46:03 +00:00
379 lines
13 KiB
Python
379 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Module implementing MainWindow.
|
|
"""
|
|
from PyQt5.QtCore import (pyqtSlot, QObject, pyqtSignal, QThreadPool,
|
|
QRunnable, QCoreApplication)
|
|
from PyQt5.QtWidgets import (QMainWindow, QProgressBar, QDialogButtonBox,
|
|
QFileDialog, QMessageBox, QApplication)
|
|
|
|
from .Ui_mainwindow import Ui_MainWindow
|
|
from .mplwidget import MplWidget
|
|
from pydsf import Experiment, PlotResults
|
|
import os
|
|
|
|
# Import available instruments
|
|
try:
|
|
from instruments.analytikJenaqTower2 import AnalytikJenaqTower2
|
|
except ImportError as err:
|
|
raise ImportError('Error while loading instrument plugins:', err)
|
|
|
|
VERSION = "1.0"
|
|
_translate = QCoreApplication.translate
|
|
|
|
|
|
class WorkerSignals(QObject):
|
|
|
|
finished = pyqtSignal()
|
|
|
|
|
|
class Worker(QRunnable):
|
|
finished = pyqtSignal(int)
|
|
|
|
def __init__(self, owner):
|
|
super().__init__()
|
|
self.exp = None
|
|
self.owner = owner
|
|
self.signals = WorkerSignals()
|
|
|
|
def run(self):
|
|
c_lower = None
|
|
c_upper = None
|
|
cbar_range = None
|
|
signal_threshold = None
|
|
if self.owner.groupBox_cutoff.isChecked():
|
|
c_lower = self.owner.doubleSpinBox_lower.value()
|
|
c_upper = self.owner.doubleSpinBox_upper.value()
|
|
if self.owner.groupBox_cbar.isChecked():
|
|
cbar_range = (self.owner.doubleSpinBox_cbar_start.value(),
|
|
self.owner.doubleSpinBox_cbar_end.value())
|
|
if self.owner.groupBox_signal_threshold.isChecked():
|
|
signal_threshold = self.owner.spinBox_signal_threshold.value()
|
|
|
|
items = (self.owner.listWidget_data.item(i)
|
|
for i in range(self.owner.listWidget_data.count()))
|
|
|
|
files = []
|
|
for item in items:
|
|
files.append(item.text())
|
|
self.exp = Experiment(instrument=self.owner.instrument,
|
|
files=files,
|
|
t1=self.owner.doubleSpinBox_tmin.value(),
|
|
t2=self.owner.doubleSpinBox_tmax.value(),
|
|
dt=self.owner.doubleSpinBox_dt.value(),
|
|
cols=12,
|
|
rows=8,
|
|
cutoff_low=c_lower,
|
|
cutoff_high=c_upper,
|
|
signal_threshold=signal_threshold,
|
|
color_range=cbar_range)
|
|
self.exp.analyze()
|
|
self.signals.finished.emit()
|
|
|
|
|
|
class TaskSignals(QObject):
|
|
finished = pyqtSignal(list)
|
|
|
|
|
|
class Tasks(QObject):
|
|
|
|
def __init__(self):
|
|
super(Tasks, self).__init__()
|
|
|
|
self.pool = QThreadPool()
|
|
self.pool.setMaxThreadCount(1)
|
|
self.tasks = []
|
|
self.data = []
|
|
self.signals = TaskSignals()
|
|
|
|
def add_task(self, task):
|
|
self.tasks.append(task)
|
|
|
|
def remove_task(self, task):
|
|
self.tasks.remove(task)
|
|
|
|
def clear(self):
|
|
self.tasks.clear()
|
|
|
|
def get_data(self):
|
|
self.pool.waitForDone()
|
|
return self.data
|
|
|
|
def start(self):
|
|
for task in self.tasks:
|
|
self.pool.start(task)
|
|
self.pool.waitForDone()
|
|
|
|
for task in self.tasks:
|
|
self.data.append(task.exp)
|
|
self.remove_task(task)
|
|
|
|
self.signals.finished.emit(self.data)
|
|
|
|
|
|
class MainWindow(QMainWindow, Ui_MainWindow):
|
|
|
|
"""
|
|
Class documentation goes here.
|
|
"""
|
|
|
|
def __init__(self, parent=None):
|
|
"""
|
|
Constructor
|
|
|
|
@param parent reference to the parent widget (QWidget)
|
|
"""
|
|
QMainWindow.__init__(self, parent)
|
|
self.setupUi(self)
|
|
self.progressBar = QProgressBar()
|
|
self.progressBar.setMaximum(0)
|
|
self.progressBar.setMaximum(0)
|
|
self.progressBar.setEnabled(False)
|
|
self.statusBar.addPermanentWidget(self.progressBar)
|
|
self.statusBar.showMessage(_translate("MainWindow",
|
|
"Welcome to PyDSF"))
|
|
self.buttonBox_process.addButton(
|
|
_translate("MainWindow", "&Start Processing"),
|
|
QDialogButtonBox.AcceptRole)
|
|
self.tasks = Tasks()
|
|
self.tasks.signals.finished.connect(self.on_processing_finished)
|
|
self.worker = None
|
|
self.outputPath = None
|
|
self.instrument = None
|
|
self.populateInstrumentList()
|
|
|
|
def populateInstrumentList(self):
|
|
self.instruments = [AnalytikJenaqTower2()]
|
|
for i in range(len(self.instruments)):
|
|
instrument = self.instruments[i]
|
|
self.comboBox_instrument.setItemText(i, instrument.name)
|
|
|
|
def getInstrumentFromName(self, name):
|
|
for instrument in self.instruments:
|
|
if instrument.name == name:
|
|
return instrument
|
|
raise IndexError("Requested instrument not")
|
|
|
|
def getSelectedInstrument(self):
|
|
name = str(self.comboBox_instrument.currentText())
|
|
instrument = self.getInstrumentFromName(name)
|
|
return instrument
|
|
|
|
@pyqtSlot("QAbstractButton*")
|
|
def on_buttonBox_open_reset_clicked(self, button):
|
|
"""
|
|
Triggered when either the open or reset button in the data file
|
|
dialog is clicked. Spawns an open file dialog or resets the listview.
|
|
"""
|
|
if button == self.buttonBox_open_reset.button(QDialogButtonBox.Open):
|
|
filenames = QFileDialog.getOpenFileNames(
|
|
self, _translate("MainWindow", "Open data file"), '',
|
|
_translate("MainWindow", "Text files (*.txt *.csv)"))
|
|
self.listWidget_data.addItems(filenames[0])
|
|
if self.listWidget_data.count() > 1:
|
|
self.groupBox_replicates.setChecked(True)
|
|
self.radioButton_rep_files.setEnabled(True)
|
|
elif button == self.buttonBox_open_reset.button(
|
|
QDialogButtonBox.Reset):
|
|
self.listWidget_data.clear()
|
|
|
|
@pyqtSlot("QAbstractButton*")
|
|
def on_buttonBox_output_clicked(self, button):
|
|
if button == self.buttonBox_output.button(QDialogButtonBox.Open):
|
|
caption = _translate('MainWindow', 'Choose output path')
|
|
path = QFileDialog.getExistingDirectory(
|
|
parent=self,
|
|
caption=caption,
|
|
options=QFileDialog.ShowDirsOnly)
|
|
self.lineEdit_output.setText(path.strip())
|
|
|
|
@pyqtSlot("QString")
|
|
def on_comboBox_instrument_currentIndexChanged(self, p0):
|
|
"""
|
|
Triggered when another instrument is selected from the combobox.
|
|
"""
|
|
self.instrument = self.getInstrumentFromName(p0)
|
|
if self.instrument.providesTempRange:
|
|
self.groupBox_temp.setEnabled(False)
|
|
else:
|
|
self.groupBox_temp.setEnabled(True)
|
|
|
|
def generate_plot_tab(self, name):
|
|
tab = MplWidget(parent=self.tabWidget)
|
|
tab.setObjectName(name)
|
|
return tab
|
|
|
|
def remove_plate_tabs(self):
|
|
for i in range(self.tabWidget.count()):
|
|
try:
|
|
widget = self.tabWidget.widget(i)
|
|
widget.deleteLater()
|
|
except IndexError:
|
|
pass
|
|
self.tabWidget.clear()
|
|
|
|
def generate_plate_tabs(self, plate):
|
|
plotter = PlotResults()
|
|
|
|
if plate.id != 'average':
|
|
tab = self.generate_plot_tab("tab_heatmap_{}".format(plate.id))
|
|
title = _translate("MainWindow", "Heatmap #")
|
|
self.tabWidget.addTab(tab, title + str(plate.id))
|
|
plotter.plot_tm_heatmap_single(plate, tab)
|
|
if self.checkBox_saveplots.isChecked():
|
|
tab.canvas.save(
|
|
"{}/heatmap_{}.svg".format(self.outputPath, plate.id))
|
|
|
|
tab = self.generate_plot_tab("tab_raw_{}".format(plate.id))
|
|
title = _translate("MainWindow", "Raw Data #")
|
|
self.tabWidget.addTab(tab, title + str(plate.id))
|
|
plotter.plot_raw(plate, tab)
|
|
if self.checkBox_saveplots.isChecked():
|
|
tab.canvas.save(
|
|
"{}/raw_{}.svg".format(self.outputPath, plate.id))
|
|
|
|
tab = self.generate_plot_tab("tab_derivative_{}".format(plate.id))
|
|
title = _translate("MainWindow", "Derivatives #")
|
|
self.tabWidget.addTab(tab, title + str(plate.id))
|
|
plotter.plot_derivative(plate, tab)
|
|
if self.checkBox_saveplots.isChecked():
|
|
tab.canvas.save(
|
|
"{}/derivatives_{}.svg".format(self.outputPath, plate.id))
|
|
else:
|
|
tab = self.generate_plot_tab("tab_heatmap_{}".format(plate.id))
|
|
title = _translate("MainWindow", "Heatmap ")
|
|
self.tabWidget.addTab(tab, title + str(plate.id))
|
|
plotter.plot_tm_heatmap_single(plate, tab)
|
|
if self.checkBox_saveplots.isChecked():
|
|
tab.canvas.save(
|
|
"{}/heatmap_{}.svg".format(self.outputPath, plate.id))
|
|
|
|
@pyqtSlot()
|
|
def on_buttonBox_process_accepted(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
|
|
self.instrument = self.getSelectedInstrument()
|
|
|
|
if self.listWidget_data.count() < 1:
|
|
QMessageBox.critical(
|
|
self, _translate("MainWindow", "Error"),
|
|
_translate("MainWindow", "No data file loaded!"),
|
|
QMessageBox.Close, QMessageBox.Close)
|
|
return
|
|
if (self.groupBox_output.isChecked() and
|
|
self.lineEdit_output.text().strip() == ''):
|
|
QMessageBox.critical(
|
|
self, _translate("MainWindow", "Error"),
|
|
_translate("MainWindow", "No output path set!"),
|
|
QMessageBox.Close, QMessageBox.Close)
|
|
return
|
|
elif (self.groupBox_output.isChecked() and
|
|
self.lineEdit_output.text().strip() != ''):
|
|
path = self.lineEdit_output.text().strip()
|
|
if os.path.isdir(path):
|
|
self.outputPath = self.lineEdit_output.text().strip()
|
|
else:
|
|
QMessageBox.critical(
|
|
self, _translate("MainWindow", "Error"),
|
|
_translate("MainWindow", "Output path does not exist!"),
|
|
QMessageBox.Close, QMessageBox.Close)
|
|
|
|
if self.spinBox_signal_threshold.value(
|
|
) == 0 and self.groupBox_signal_threshold.isChecked():
|
|
QMessageBox.warning(
|
|
self, _translate("MainWindow", "Warning"), _translate(
|
|
"MainWindow",
|
|
"Signal threshold is currently set to zero."),
|
|
QMessageBox.Ok, QMessageBox.Ok)
|
|
|
|
self.remove_plate_tabs()
|
|
self.progressBar.setEnabled(True)
|
|
self.statusBar.showMessage(_translate("MainWindow", "Processing..."))
|
|
|
|
self.worker = Worker(self)
|
|
self.tasks.add_task(self.worker)
|
|
self.tasks.start()
|
|
|
|
@pyqtSlot()
|
|
def on_processing_finished(self):
|
|
# Clear all jobs from task list
|
|
# self.tasks.clear()
|
|
# For now, only read the first entry
|
|
exp = self.tasks.data[0]
|
|
|
|
for i in range(len(self.worker.exp.plates)):
|
|
|
|
plate = exp.plates[i]
|
|
self.generate_plate_tabs(plate)
|
|
|
|
if exp.avg_plate:
|
|
|
|
plate = exp.avg_plate
|
|
self.generate_plate_tabs(plate)
|
|
|
|
if self.groupBox_output.isChecked():
|
|
if self.checkBox_savetables.isChecked():
|
|
for plate in exp.plates:
|
|
plate.write_tm_table(
|
|
'{}/plate_{}_tm.csv'.format(self.outputPath,
|
|
str(plate.id)))
|
|
plate.write_data_table(
|
|
'{}/plate_{}_dI_dT.csv'.format(self.outputPath,
|
|
str(plate.id)),
|
|
dataType='derivative')
|
|
plate.write_data_table(
|
|
'{}/plate_{}_filtered_data.csv'.format(self.outputPath,
|
|
str(plate.id)),
|
|
dataType='filtered')
|
|
plate.write_data_table('{}/plate_{}_raw_data.csv'.format(
|
|
self.outputPath, str(plate.id)))
|
|
|
|
if exp.avg_plate:
|
|
exp.avg_plate.write_tm_table(
|
|
'{}/plate_{}_tm_avg.csv'.format(
|
|
self.outputPath,
|
|
str(self.worker.exp.avg_plate.id)),
|
|
avg=True)
|
|
|
|
self.progressBar.setEnabled(False)
|
|
self.statusBar.showMessage(_translate("MainWindow", "Finished!"))
|
|
|
|
@pyqtSlot()
|
|
def on_buttonBox_process_rejected(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
QApplication.quit()
|
|
|
|
pyqtSlot()
|
|
|
|
def on_actionQuit_triggered(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
QApplication.quit()
|
|
|
|
@pyqtSlot("bool")
|
|
def on_groupBox_cutoff_toggled(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
self.doubleSpinBox_upper.setValue(self.doubleSpinBox_tmax.value())
|
|
self.doubleSpinBox_lower.setValue(self.doubleSpinBox_tmin.value())
|
|
|
|
@pyqtSlot()
|
|
def on_actionAbout_triggered(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
# TODO: not implemented yet
|
|
raise NotImplementedError
|
|
|
|
@pyqtSlot()
|
|
def on_actionAbout_Qt_triggered(self):
|
|
"""
|
|
Slot documentation goes here.
|
|
"""
|
|
QApplication.aboutQt()
|