Source code for aca_view.widgets.data_source_dialog
from cxotime import CxoTime
from PyQt5 import QtCore as QtC
from PyQt5 import QtGui as QtG
from PyQt5 import QtWidgets as QtW
from aca_view.data.config import validate_config
[docs]
class DateTimeEdit(QtW.QWidget):
"""
Widget to get date-time from the user, using a standard date-time widget or a string.
"""
date_changed = QtC.pyqtSignal(str)
def __init__(self):
QtW.QWidget.__init__(self)
self.setLayout(QtW.QHBoxLayout())
self.calendar_edit = QtW.QDateTimeEdit(self)
self.calendar_edit.setCalendarPopup(True)
dt = CxoTime(self.calendar_edit.dateTime().toString("yyyy-MM-dd hh:mm:ss"))
self.text_edit = QtW.QLineEdit(self)
self.text_edit.setText(dt.date)
icon = QtG.QIcon(":toolbar/normal/calendar.png")
self.calendar_label = QtW.QLabel()
self.calendar_label.setPixmap(icon.pixmap(24, 24))
self.calendar_label.setToolTip("Show calendar")
icon = QtG.QIcon(":toolbar/normal/text.png")
self.text_label = QtW.QLabel()
self.text_label.setPixmap(icon.pixmap(24, 24))
self.text_label.setToolTip("Text format")
# set the minimum sizes so the window size does not change when one toggles between them
# and so the data fits in the text box
fm = self.text_edit.fontMetrics()
w = fm.boundingRect(self.text_edit.text()).width() + 10
s1 = self.text_edit.minimumSizeHint()
s2 = self.calendar_edit.minimumSizeHint()
h = max(s1.height(), s2.height()) + 2
w = max(w, s1.width(), s2.width()) + 2
self.text_edit.setMinimumSize(w, h)
self.calendar_edit.setMinimumSize(w, h)
self.layout().addWidget(self.calendar_edit)
self.layout().addWidget(self.text_edit)
self.layout().addWidget(self.text_label)
self.layout().addWidget(self.calendar_label)
self.calendar_edit.hide()
self.text_label.hide()
# I set connections at the end, after things are initialized, so they do not trigger early
self.text_edit.textChanged.connect(self._text_edit_changed)
self.calendar_edit.dateTimeChanged.connect(self._calendar_edit_changed)
def mousePressEvent(self, _):
if self.calendar_edit.isHidden():
self.text_edit.hide()
self.calendar_label.hide()
self.calendar_edit.show()
self.text_label.show()
else:
self.calendar_edit.hide()
self.text_label.hide()
self.text_edit.show()
self.calendar_label.show()
def _text_edit_changed(self, date_time_str):
self.date_changed.emit(date_time_str)
def _calendar_edit_changed(self, dt):
date_time_str = CxoTime(dt.toString("yyyy-MM-dd hh:mm:ss")).date
self.date_changed.emit(date_time_str)
[docs]
class DataSourceDialog(QtW.QDialog):
"""
Dialog to configure data fetching.
"""
def __init__(self, defaults=None, verbose=False): # noqa: PLR0915
QtW.QDialog.__init__(self)
self.verbose = verbose
self.state = {
"real_time": False,
"mica": True,
"maude": True,
"cheta_sources": ["cxc", "maude"],
"maude_channel": "FLIGHT",
}
self.setLayout(QtW.QVBoxLayout())
self.tab = QtW.QTabWidget(self)
self.obsid_widget = QtW.QLineEdit()
self.obsid_widget.textChanged.connect(self._set_obsid)
self.start_widget = DateTimeEdit()
self.start_widget.date_changed.connect(self._set_start)
self.stop_widget = DateTimeEdit()
self.stop_widget.date_changed.connect(self._set_stop)
self.real_time = QtW.QRadioButton("Real-time")
self.archival = QtW.QRadioButton("Archival")
self.mode_group = QtW.QButtonGroup()
self.mode_group.addButton(self.real_time)
self.mode_group.addButton(self.archival)
self.mode_group.buttonClicked.connect(self._set_mode)
self.archival.setChecked(not self.state["real_time"])
self.maude = QtW.QCheckBox("MAUDE")
self.maude.toggled.connect(self._enable_maude)
self.maude.setChecked(self.state["maude"])
self.mica = QtW.QCheckBox("Mica")
self.mica.toggled.connect(self._enable_mica)
self.mica.setChecked(self.state["mica"])
self.cheta_maude = QtW.QCheckBox("MAUDE")
self.cheta_maude.toggled.connect(self._enable_cheta_maude)
self.cheta_maude.setChecked("maude" in self.state["cheta_sources"])
self.cheta_cxc = QtW.QCheckBox("CXC")
self.cheta_cxc.toggled.connect(self._enable_cheta_cxc)
self.cheta_cxc.setChecked("cxc" in self.state["cheta_sources"])
self.maude_channel_flight = QtW.QRadioButton("FLIGHT")
self.maude_channel_flight.setChecked(self.state["maude_channel"] == "FLIGHT")
self.maude_channel_asvt = QtW.QRadioButton("ASVT")
self.maude_channel_asvt.setChecked(self.state["maude_channel"] == "ASVT")
self.channel_group = QtW.QButtonGroup()
self.channel_group.addButton(self.maude_channel_flight)
self.channel_group.addButton(self.maude_channel_asvt)
self.channel_group.buttonClicked.connect(self._set_maude_channel)
self.filenames_widget = QtW.QListWidget()
tab_1 = QtW.QWidget()
self.tab.addTab(tab_1, "Main")
tab_2 = QtW.QWidget()
self.tab.addTab(tab_2, "filenames")
layout = QtW.QGridLayout()
layout_2 = QtW.QGridLayout()
msid_group = QtW.QGroupBox("OBSID")
msid_layout = QtW.QHBoxLayout()
msid_layout.addWidget(self.obsid_widget)
msid_group.setLayout(msid_layout)
time_range_group = QtW.QGroupBox("TimeRange")
time_range_layout = QtW.QGridLayout()
time_range_layout.addWidget(QtW.QLabel("start"), 0, 0)
time_range_layout.addWidget(self.start_widget, 0, 1)
time_range_layout.addWidget(QtW.QLabel("stop"), 1, 0)
time_range_layout.addWidget(self.stop_widget, 1, 1)
time_range_group.setLayout(time_range_layout)
mode_button_group = QtW.QGroupBox("Mode")
button_layout = QtW.QHBoxLayout()
button_layout.addWidget(self.archival)
button_layout.addWidget(self.real_time)
mode_button_group.setLayout(button_layout)
source_button_group = QtW.QGroupBox("Image Telemetry Source")
button_layout = QtW.QHBoxLayout()
button_layout.addWidget(self.maude)
button_layout.addWidget(self.mica)
source_button_group.setLayout(button_layout)
cheta_source_button_group = QtW.QGroupBox('Cheta Source ("other" telemetry)')
button_layout = QtW.QHBoxLayout()
button_layout.addWidget(self.cheta_cxc)
button_layout.addWidget(self.cheta_maude)
cheta_source_button_group.setLayout(button_layout)
maude_channel_button_group = QtW.QGroupBox("MAUDE channel")
button_layout = QtW.QHBoxLayout()
button_layout.addWidget(self.maude_channel_flight)
button_layout.addWidget(self.maude_channel_asvt)
maude_channel_button_group.setLayout(button_layout)
layout_2.addWidget(self.filenames_widget)
choose_files = QtW.QPushButton("Choose")
choose_files.clicked.connect(self._choose_files)
layout_2.addWidget(choose_files)
layout.addWidget(mode_button_group)
layout.addWidget(msid_group)
layout.addWidget(time_range_group)
layout.addWidget(source_button_group)
layout.addWidget(cheta_source_button_group)
layout.addWidget(maude_channel_button_group)
tab_2.setLayout(layout_2)
tab_1.setLayout(layout)
button_box = QtW.QDialogButtonBox(
QtW.QDialogButtonBox.Ok | QtW.QDialogButtonBox.Cancel
)
button_box.accepted.connect(self.accept)
button_box.rejected.connect(self.reject)
self.filenames_widget.doubleClicked.connect(self._choose_files)
self.filenames_widget.clicked.connect(self._choose_files)
self.layout().addWidget(self.tab)
self.layout().addWidget(button_box)
self.feedback_widget = None
def _choose_files(self):
file_dialog = QtW.QFileDialog()
if file_dialog.exec():
filenames = file_dialog.selectedFiles()
for filename in filenames:
self.filenames_widget.addItem(filename)
self.state["filenames"] = [
self.filenames_widget.item(i).text()
for i in range(self.filenames_widget.count())
]
def _set_mode(self, mode):
archival = mode == self.archival
self.state["real_time"] = not archival
self.stop_widget.setEnabled(archival)
self.mica.setEnabled(archival)
self.maude.setEnabled(archival)
self.obsid_widget.setEnabled(archival)
self.cheta_cxc.setEnabled(archival)
self.cheta_maude.setEnabled(archival)
self.tab.setTabEnabled(1, archival)
def _set_maude_channel(self, channel):
flight = channel == self.maude_channel_flight
self.state["maude_channel"] = "FLIGHT" if flight else "ASVT"
def _set_obsid(self, obsid):
self.state["obsid"] = obsid.strip() if obsid else None
def _set_start(self, dt):
self.state["start"] = dt
def _set_stop(self, dt):
self.state["stop"] = dt
def _enable_mica(self, enabled):
self.state["mica"] = enabled
def _enable_maude(self, enabled):
self.state["maude"] = enabled
def _enable_cheta_cxc(self, enabled):
if not enabled and "cxc" in self.state["cheta_sources"]:
self.state["cheta_sources"].remove("cxc")
if enabled and "cxc" not in self.state["cheta_sources"]:
self.state["cheta_sources"] += ["cxc"]
self.state["cheta_sources"].sort()
def _enable_cheta_maude(self, enabled):
if not enabled and "maude" in self.state["cheta_sources"]:
self.state["cheta_sources"].remove("maude")
if enabled and "maude" not in self.state["cheta_sources"]:
self.state["cheta_sources"] += ["maude"]
self.state["cheta_sources"].sort()
def _get_state(self):
result = self.state.copy()
# modify it depending on what is enabled
return result
def accept(self):
try:
import pprint
if self.verbose:
pprint.pprint(self._get_state())
self.state = validate_config(**self._get_state())
if self.verbose:
print("Final accepted state:")
pprint.pprint(self.state)
QtW.QDialog.accept(self)
except Exception as e:
if self.feedback_widget is None:
self.feedback_widget = QtW.QTextEdit("")
self.layout().insertWidget(1, self.feedback_widget)
self.feedback_widget.setText(f"Invalid Data: {e}")
if __name__ == "__main__":
from aca_view.tests.utils import qt
with qt():
app = DataSourceDialog(verbose=True)
app.show()