import pprint
from cheta import fetch
from cxotime import CxoTime
from jinja2 import Template
from PyQt5 import QtCore as QtC
from PyQt5 import QtGui as QtG
from PyQt5 import QtWidgets as QtW
from aca_view.config import COMMAND_LINE_OPTIONS
from aca_view.data.config import ConfigError, 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)
self.text_edit = QtW.QLineEdit(self)
self.text_edit.setText("")
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
defaults = {
"real_time": False,
"mica": True,
"maude": True,
"cheta_sources": ["cxc", "maude"],
"maude_channel": "FLIGHT",
"ska_data_sources": ["local"],
}
defaults.update(COMMAND_LINE_OPTIONS.get("data_service", {}))
mica = defaults["mica"]
maude = defaults["maude"]
if not mica and not maude:
maude = True
mica = True
self.state = {
"real_time": defaults["real_time"],
"mica": mica,
"maude": maude,
# the logic here is different from the one used in the command line
# in the command line, if one cheta source (maude or cxc) is explicitly requested
# and the other is not, then only the requested one is used
# if none are requested, then it defaults to fetch.data_source.sources()
# here we default to fetch.data_source.sources() when the dialog is created.
"cheta_sources": fetch.data_source.sources(),
"maude_channel": defaults["maude_channel"],
"ska_data_sources": defaults["ska_data_sources"],
}
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")
tab_3 = QtW.QWidget()
self.tab.addTab(tab_3, "Data Sources")
layout = QtW.QGridLayout()
layout_2 = QtW.QGridLayout()
layout_3 = 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("Non-image 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_3.addWidget(source_button_group)
layout_3.addWidget(cheta_source_button_group)
layout_3.addWidget(maude_channel_button_group)
tab_3.setLayout(layout_3)
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.itemClicked.connect(self._filename_clicked)
self.layout().addWidget(self.tab)
self.layout().addWidget(button_box)
self.feedback_widget_ = None
def _filename_clicked(self, item):
"""
Handle the click on a filename in the list. Currently just a menu to delete the entry.
"""
filename = item.text()
if filename:
menu = QtW.QMenu(self)
delete_action = QtW.QAction("Delete", self)
delete_action.triggered.connect(
lambda: self.filenames_widget.takeItem(self.filenames_widget.row(item))
)
menu.addAction(delete_action)
menu.exec(QtG.QCursor.pos())
def _choose_files(self):
file_dialog = QtW.QFileDialog()
file_dialog.setFileMode(QtW.QFileDialog.ExistingFiles)
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"] = list(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
@property
def feedback_widget(self):
if self.feedback_widget_ is None:
self.feedback_widget_ = QtW.QTextEdit("")
self.tab.addTab(self.feedback_widget_, "Errors")
return self.feedback_widget_
def accept(self):
try:
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 ConfigError as e:
template = Template(
"""
<center> <h3>Data fetching cannot be configured </h3> </center>
<ul>
{% for error in errors %}
<li>{{ error }}</li>
{% endfor %}
</ul>
"""
)
self.feedback_widget.setText(template.render(errors=e.errors))
self.tab.setCurrentWidget(self.feedback_widget)
except Exception as e:
self.feedback_widget.setText(f"Invalid Data: {e}")
self.tab.setCurrentWidget(self.feedback_widget)
def show_error(self, *errors, message="", title="Configuration Error"):
template = Template(
"""
<center> <h3> {{ title }} </h3> </center>
{{ message }}
<ul>
{% for error in errors %}
<li>{{ error }}</li>
{% endfor %}
</ul>
"""
)
self.feedback_widget.setText(
template.render(title=title, message=message, errors=errors)
)
self.tab.setCurrentWidget(self.feedback_widget)
if __name__ == "__main__":
from aca_view.tests.utils import qt
with qt():
app = DataSourceDialog(verbose=True)
app.show()