"""
The Calibrations view of the PypeIt Dashboard (Stage 3).
The drill-down companion to the Status view: scope selectors (C2), a
path-aware step-button row (C3/C4/C12/C13), and a detail panel (C5–C9, C11)
for the selected step — metrics, input files, the output viewer, QA files,
and a per-slit/order table. It launches the existing inspection tools as
subprocesses via the :class:`~pypeit.dashboard.launcher.Launcher` (C8/C14/C15)
and stays thin: data via the model, colors via the palette, commands via
``inspect``.
Generated by JXP and Claude.
"""
from pathlib import Path
from qtpy.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QComboBox, QPushButton, QTableWidget,
QTableWidgetItem, QHeaderView, QListWidget,
QListWidgetItem, QScrollArea, QGroupBox,
QAbstractItemView, QMessageBox)
from qtpy.QtCore import Qt
from qtpy.QtGui import QColor, QPalette
from pypeit import log
from pypeit.dashboard import palette
from pypeit.dashboard import inspect as dash_inspect
from pypeit.dashboard.view.qa_dialog import QaImageDialog
# Steps that carry per-slit/order detail (C11).
_PER_SLIT_STEPS = ('slits', 'wv_calib', 'tilts', 'flats')
_MAGENTA = '#D81B60' # selected-step ring (design)
[docs]
def _clear_layout(layout):
"""
Remove and delete every item from a Qt layout.
Generated by JXP and Claude.
Args:
layout (``QLayout``): The layout to empty.
Returns:
None.
"""
while layout.count():
item = layout.takeAt(0)
widget = item.widget()
if widget is not None:
# Reparent out of the visible tree immediately (deleteLater is
# deferred to the event loop, which would otherwise leave the old
# widgets overlapping a freshly rebuilt panel).
widget.setParent(None)
widget.deleteLater()
else:
child = item.layout()
if child is not None:
_clear_layout(child)
[docs]
def _text_on(hexcolor):
"""
Pick a readable text color (black/white) for a background hex color.
Generated by JXP and Claude.
Args:
hexcolor (:obj:`str`): Background color, e.g. ``#2E7D32``.
Returns:
str: ``'#000000'`` or ``'#FFFFFF'``.
"""
c = QColor(hexcolor)
luminance = 0.299 * c.redF() + 0.587 * c.greenF() + 0.114 * c.blueF()
return '#000000' if luminance > 0.6 else '#FFFFFF'
[docs]
class CalibrationsView(QWidget):
"""
The Calibrations tab: the per-(group, detector) calibration drill-down.
Generated by JXP and Claude.
Args:
model (:class:`~pypeit.dashboard.model.DashboardModel`):
The reduction-state model.
launcher (:class:`~pypeit.dashboard.launcher.Launcher`, optional):
Used to launch inspection tools and (re)build runs; if ``None``,
the controls are built but do nothing (e.g. in headless tests).
run_lock (:class:`~pypeit.dashboard.runlock.RunLock`, optional):
The single-run lock (design X1); the (Re)Build control is disabled
while it is locked.
on_run_finished (callable, optional):
Called as ``on_run_finished(code)`` when a (Re)Build run ends, so
the window can refresh the state (design C15).
parent (:obj:`QWidget`, optional):
The parent widget.
"""
def __init__(self, model, launcher=None, run_lock=None,
on_run_finished=None, parent=None):
super().__init__(parent=parent)
self._model = model
self._launcher = launcher
self._run_lock = run_lock
self._on_run_finished = on_run_finished
self._theme = 'light'
self._scope = None # (group, det)
self._selected_step = None
self._group_combo = None
self._det_combo = None
self._button_row = None # QHBoxLayout holding the step buttons
self._step_buttons = {} # step -> QPushButton
self._detail = None # QVBoxLayout for the detail panel
self._table = None # the step-button-row's status lookup
# The current detail panel's (Re)Build button + whether it has a
# command, so the lock can enable/disable it without a full rebuild.
self._rebuild_button = None
self._rebuild_available = False
self._rebuild_step = None # step the current (Re)Build button targets
self._outer = QVBoxLayout(self)
self._build()
# -- public API ------------------------------------------------------
[docs]
def set_model(self, model):
"""
Swap in a new model and rebuild.
Generated by JXP and Claude.
Args:
model (:class:`~pypeit.dashboard.model.DashboardModel`):
The new model.
Returns:
None.
"""
self._model = model
self._scope = None
self._selected_step = None
_clear_layout(self._outer)
self._build()
[docs]
def refresh(self, model):
"""
Swap in a new model but **preserve** the current scope and selected
step (used after a (Re)Build completes, so the rebuilt step stays
selected rather than resetting to the first step; C15, Round-2 #2).
Generated by JXP and Claude.
Args:
model (:class:`~pypeit.dashboard.model.DashboardModel`):
The freshly-acquired model.
Returns:
None.
"""
prev_scope = self._scope
prev_step = self._selected_step
self._model = model
self._scope = None
# Keep the desired selection; _rebuild_button_row honors it if the
# step is still present.
self._selected_step = prev_step
_clear_layout(self._outer)
self._build()
# _build scopes to the first pair; restore the previous scope if it
# still exists in the refreshed state.
if prev_scope is not None and prev_scope in model.calib_det_pairs():
self._set_scope(prev_scope)
# -- build -----------------------------------------------------------
[docs]
def _build(self):
"""
Build the view (edge message or full drill-down).
Generated by JXP and Claude.
Returns:
None.
"""
self._theme = self._detect_theme()
if not self._model.is_started():
msg = QLabel('No calibration state to inspect yet.')
msg.setAlignment(Qt.AlignCenter)
msg.setStyleSheet('color: grey; font-style: italic; '
'font-size: 16px;')
self._outer.addStretch(1)
self._outer.addWidget(msg)
self._outer.addStretch(1)
return
self._build_scope_toolbar()
# The step-button row (rebuilt on scope change).
self._button_row = QHBoxLayout()
row_wrap = QHBoxLayout()
row_wrap.addLayout(self._button_row)
row_wrap.addStretch(1)
self._outer.addLayout(row_wrap)
# The detail panel (rebuilt on step selection), in a scroll area.
self._detail = QVBoxLayout()
detail_container = QWidget()
detail_container.setLayout(self._detail)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setWidget(detail_container)
self._outer.addWidget(scroll, stretch=1)
pairs = self._model.calib_det_pairs()
if pairs:
self._set_scope(pairs[0])
[docs]
def _detect_theme(self):
"""
Detect light vs dark from the widget palette.
Generated by JXP and Claude.
Returns:
str: ``'dark'`` or ``'light'``.
"""
color = self.palette().color(QPalette.Window)
lum = (0.299 * color.redF() + 0.587 * color.greenF()
+ 0.114 * color.blueF())
return 'dark' if lum < 0.5 else 'light'
# -- scope handling --------------------------------------------------
[docs]
def _reload_det_combo(self, select=None):
"""
Repopulate the detector drop-down for the current group.
Generated by JXP and Claude.
Args:
select: Detector to select after repopulating, if present.
Returns:
None.
"""
group = self._group_combo.currentData()
dets = [d for g, d in self._model.calib_det_pairs() if g == group]
self._det_combo.blockSignals(True)
self._det_combo.clear()
for d in dets:
self._det_combo.addItem(self._model.det_name(d), d)
if select is not None:
idx = self._det_combo.findData(select)
if idx >= 0:
self._det_combo.setCurrentIndex(idx)
self._det_combo.blockSignals(False)
[docs]
def _set_scope(self, pair):
"""
Set the scoped ``(group, det)``, syncing the drop-downs and
rebuilding the button row.
Generated by JXP and Claude.
Args:
pair (:obj:`tuple`): ``(group, det)``.
Returns:
None.
"""
self._scope = pair
group, det = pair
self._group_combo.blockSignals(True)
idx = self._group_combo.findData(group)
if idx >= 0:
self._group_combo.setCurrentIndex(idx)
self._group_combo.blockSignals(False)
self._reload_det_combo(select=det)
self._rebuild_button_row()
[docs]
def _on_group_changed(self, _index):
"""
Handle a group change: reload detectors, rebuild the button row.
Generated by JXP and Claude.
Args:
_index (:obj:`int`): The new combo index (unused).
Returns:
None.
"""
self._reload_det_combo()
self._scope = (self._group_combo.currentData(),
self._det_combo.currentData())
self._rebuild_button_row()
[docs]
def _on_det_changed(self, _index):
"""
Handle a detector change: rebuild the button row.
Generated by JXP and Claude.
Args:
_index (:obj:`int`): The new combo index (unused).
Returns:
None.
"""
self._scope = (self._group_combo.currentData(),
self._det_combo.currentData())
self._rebuild_button_row()
# -- step-button row -------------------------------------------------
[docs]
def _scope_status(self):
"""
Return a ``step -> (required, status, in_pipeline)`` lookup for the
current scope.
Generated by JXP and Claude.
Returns:
dict: Per-step status tuples for the selected (group, det).
"""
group, det = self._scope
table = self._model.status_table()
mask = (table['calibration_group'] == group) \
& (table['detector'] == det)
lookup = {}
for row in table[mask].itertuples(index=False):
lookup[row.step] = (row.required, row.status, row.in_pipeline)
return lookup
[docs]
def _select_step(self, step):
"""
Select a step: re-style the buttons and rebuild the detail panel.
Generated by JXP and Claude.
Args:
step (:obj:`str`): The selected step.
Returns:
None.
"""
self._selected_step = step
self._restyle_buttons()
self._build_detail(step)
# -- detail panel ----------------------------------------------------
[docs]
def _build_detail(self, step):
"""
Rebuild the detail panel for the selected step (C5–C9, C11).
Generated by JXP and Claude.
Args:
step (:obj:`str`): The selected step.
Returns:
None.
"""
_clear_layout(self._detail)
# The detail panel is rebuilt, so the old (Re)Build button is gone.
self._rebuild_button = None
self._rebuild_available = False
group, det = self._scope
entry = self._model.step_entry(step, group, det)
heading = QLabel(f'{step}')
heading.setStyleSheet('font-size: 15px; font-weight: bold;')
self._detail.addWidget(heading)
# Metrics (C6) — entry-level; per-slit metrics are in the table below.
# Numeric values are shown to a few significant figures (Round-2 #2).
metrics = self._model.step_metrics(step, group, det)
if metrics:
text = ' '.join(f'{k}: {self._fmt(v)}'
for k, v in metrics.items())
self._detail.addWidget(QLabel(text))
# Action row (top): "Inspect output" (C8) + "(Re)Build" (C10), the
# step's two primary actions side by side (S4-Q1).
self._add_action_row(step, group, det)
# Output filename (Round-2 #3): show it for every step that has one —
# useful especially for wv_calib, which has no viewer.
self._add_output_label(entry)
# QA files (C9).
self._add_qa_files(entry)
# Per-slit/order drill-down (C11).
if step in _PER_SLIT_STEPS:
self._add_slit_table(step, group, det)
# Input files (C7) live at the **bottom**, half-width and short
# (Round-1 #3); grouped by role for flats (S3-Q16), union otherwise.
self._detail.addStretch(1)
self._add_input_files(step, entry)
[docs]
def _add_action_row(self, step, group, det):
"""
Add the action row: "Inspect output" (C8) beside "(Re)Build" (C10).
Generated by JXP and Claude.
Args:
step (:obj:`str`): The step name.
group (:obj:`int`): The calibration group.
det: The detector.
Returns:
None.
"""
row = QHBoxLayout()
row.addWidget(self._build_output_button(step, group, det))
row.addWidget(self._build_rebuild_button(step, group, det))
row.addStretch(1)
self._detail.addLayout(row)
[docs]
def _add_output_label(self, entry):
"""
Show the step's output filename (Round-2 #3), so it is visible even
when there is no viewer (e.g. ``wv_calib``).
Generated by JXP and Claude.
Args:
entry: The step's state entry (or ``None``).
Returns:
None.
"""
name = getattr(entry, 'output_file', None) if entry else None
if not name:
return
label = QLabel(f'Output: {Path(name).name}')
label.setStyleSheet('color: grey;')
# Selectable so the user can copy the filename.
label.setTextInteractionFlags(Qt.TextSelectableByMouse)
self._detail.addWidget(label)
[docs]
def set_locked(self, locked):
"""
Restyle the (Re)Build control when the run lock changes (X1, Round-3):
orange + disabled while locked, blue + enabled when idle.
Generated by JXP and Claude.
Args:
locked (:obj:`bool`): Whether a run is in progress.
Returns:
None.
"""
self._style_rebuild_button(locked)
[docs]
def _on_rebuild(self, step, group, det):
"""
Confirm and launch a (re)build of the selected step (C10, X2/X3).
Shows a clobber confirmation naming the file(s) that will be
overwritten (S4-Q4); on confirm, moves only the selected step's
output(s) aside (S4-Q3 option (a) — "delete-then-run", in code) and
launches ``pypeit_run_to_calibstep`` through the launcher, which holds
the lock and refreshes on completion (C15).
The move-aside is **crash-safe**: each output is renamed to a
``.dashboard_bak`` sibling rather than unlinked, and
:meth:`_after_rebuild` restores it if the run **fails** (so a failed
(re)build never loses the existing calibration; Round-1 #2).
Generated by JXP and Claude.
Args:
step (:obj:`str`): The step name.
group (:obj:`int`): The calibration group.
det: The detector.
Returns:
None.
"""
# Refuse if a run is already active (defensive; the button is also
# disabled while locked).
if self._run_lock is not None and self._run_lock.is_locked():
return
argv = dash_inspect.run_command(self._model, step, group, det)
if argv is None:
return
existing = self._model.step_output_files(step, group, det)
if not self._confirm_rebuild(step, existing):
return
# Move the selected step's output(s) aside so the reused
# run_to_calibstep genuinely rebuilds them, keeping a backup to restore
# if the run fails (option (a), made crash-safe; never a shell rm).
backups = []
for path in existing:
bak = path.parent / (path.name + '.dashboard_bak')
try:
if bak.exists():
bak.unlink()
path.rename(bak)
backups.append((bak, path))
except OSError as exc:
log.warning(f'Could not move {path} aside: {exc}')
if self._launcher is not None:
# Launching engages the run lock, which turns the (Re)Build button
# orange + disabled via set_locked() (Round-3 #1/#2).
self._launcher.run(
argv, description=f'(Re)building {step}',
on_finished=lambda code, s=step, b=backups:
self._after_rebuild(code, s, b))
[docs]
def _after_rebuild(self, code, step, backups):
"""
Finish a (re)build: drop the backups on success, restore them on
failure, keep the rebuilt step selected, then trigger the window
refresh (C15, Round-1 #2, Round-2 #2).
Generated by JXP and Claude.
Args:
code (:obj:`int`): The run's exit code (0 = success).
step (:obj:`str`): The step that was (re)built — kept selected so
the refreshed view shows it, not the first step.
backups (:obj:`list`): ``(backup_path, original_path)`` pairs moved
aside before the run.
Returns:
None.
"""
for bak, original in backups:
try:
if code == 0 or original.exists():
# Success (or the run wrote a fresh output anyway): the
# backup is no longer needed.
bak.unlink()
else:
# Failed and nothing was rebuilt: restore the original so
# the calibration is not lost.
bak.rename(original)
except OSError as exc:
log.warning(f'Could not finalize backup {bak}: {exc}')
# Keep the rebuilt step selected through the refresh (Round-2 #2).
self._selected_step = step
if self._on_run_finished is not None:
self._on_run_finished(code)
[docs]
def _confirm_rebuild(self, step, existing):
"""
Show the clobber confirmation dialog (X2), naming the file(s) to be
overwritten.
Generated by JXP and Claude.
Args:
step (:obj:`str`): The step name.
existing (:obj:`list`): Existing output :obj:`pathlib.Path` files.
Returns:
bool: True if the user confirmed the (re)build.
"""
if existing:
files = '\n'.join(f' • {p.name}' for p in existing)
text = (f'(Re)build "{step}" — this will overwrite the existing '
f'output file(s):\n\n{files}\n\nPreceding calibrations are '
f'reused. Continue?')
icon = QMessageBox.Warning
else:
text = (f'(Re)build "{step}"? No existing output for this step was '
f'found, so it (and any preceding steps) will be built '
f'fresh. Continue?')
icon = QMessageBox.Question
box = QMessageBox(self)
box.setIcon(icon)
box.setWindowTitle('Confirm (Re)Build')
box.setText(text)
box.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel)
box.setDefaultButton(QMessageBox.Cancel)
return box.exec_() == QMessageBox.Ok
# Steps whose viewer opens a Ginga window (vs a matplotlib/Qt plot).
_GINGA_HINT_STEPS = ('bias', 'dark', 'arc', 'tiltimg', 'slits',
'scattlight', 'wv_calib')
[docs]
def _viewer_hint(self, step):
"""
Return where a step's output viewer shows its result, for the
activity bar's finished message (Round-1 #1).
Generated by JXP and Claude.
Args:
step (:obj:`str`): The step name.
Returns:
str: ``'Ginga window'`` or ``'plot window'``.
"""
return 'Ginga window' if step in self._GINGA_HINT_STEPS \
else 'plot window'
[docs]
def _add_file_item(self, listw, path, prefix=''):
"""
Add one file row to an input-file list, stashing the path.
Generated by JXP and Claude.
Args:
listw (:obj:`QListWidget`): The list to add to.
path (:obj:`str`): The file path.
prefix (:obj:`str`, optional): Display prefix (e.g. a role tag).
Returns:
None.
"""
item = QListWidgetItem(f'{prefix}{Path(path).name}')
item.setData(Qt.UserRole, str(path))
listw.addItem(item)
[docs]
def _add_qa_files(self, entry):
"""
Add the QA-file list (C9); double-click opens the PNG full-view.
Generated by JXP and Claude.
Args:
entry: The step's state entry (or ``None``).
Returns:
None.
"""
qa_files = getattr(entry, 'qa_files', None) if entry else None
if not qa_files:
return
self._detail.addWidget(QLabel('QA files (double-click to open):'))
listw = QListWidget()
listw.setMaximumHeight(110)
for f in qa_files:
self._add_file_item(listw, f)
listw.itemDoubleClicked.connect(self._on_qa_double_clicked)
self._detail.addWidget(listw)
[docs]
def _on_qa_double_clicked(self, item):
"""
Open the double-clicked QA PNG in an image dialog.
Generated by JXP and Claude.
Args:
item (:obj:`QListWidgetItem`): The clicked item.
Returns:
None.
"""
path = Path(item.data(Qt.UserRole))
if not path.is_absolute() and not path.exists():
# QA PNGs live under <redux_dir>/QA/PNGs/.
candidate = self._model.redux_dir / 'QA' / 'PNGs' / path.name
if candidate.exists():
path = candidate
dialog = QaImageDialog(str(path), parent=self)
dialog.show()
[docs]
def _add_slit_table(self, step, group, det):
"""
Add the per-slit/order table (C11). For flats the table has
per-correction mean/RMS columns (S3-Q14).
Generated by JXP and Claude.
Args:
step (:obj:`str`): The step name.
group (:obj:`int`): The calibration group.
det: The detector.
Returns:
None.
"""
rows = self._model.slit_table(step, group, det)
if not rows:
return
# For the Echelle pipeline these rows are spectral orders, so label
# them "Order" rather than "Slit" (S3b-Q1).
is_echelle = self._model.header_info is not None \
and self._model.header_info.path == 'Echelle'
unit = 'order' if is_echelle else 'slit'
self._detail.addWidget(QLabel(f'Per-{unit} ({len(rows)}):'))
# Build columns: Slit/Order, Status, then step-specific metric columns.
if step == 'slits':
metric_cols = ['center', 'slitord_id']
elif step in ('wv_calib', 'tilts'):
metric_cols = ['rms']
else: # flats: a mean/rms pair per present correction
corrections = self._model.step_metrics(
step, group, det).get('corrections', [])
metric_cols = []
for corr in corrections:
metric_cols += [f'{corr} mean', f'{corr} rms']
columns = [unit.capitalize(), 'Status'] + metric_cols
table = QTableWidget()
table.setColumnCount(len(columns))
table.setHorizontalHeaderLabels(columns)
table.setRowCount(len(rows))
table.verticalHeader().setVisible(False)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionMode(QAbstractItemView.NoSelection)
table.setAlternatingRowColors(True)
# Neutral selection fill, consistent with the science tables (Round-1 #2).
table.setStyleSheet(palette.selection_style(self._theme))
table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
for r, row in enumerate(rows):
slit_item = QTableWidgetItem(str(row['slit']))
slit_item.setTextAlignment(Qt.AlignCenter)
table.setItem(r, 0, slit_item)
style = palette.slit_style(row['status'], theme=self._theme)
status_item = QTableWidgetItem(f'{style.glyph} {style.label}')
status_item.setTextAlignment(Qt.AlignCenter)
if style.category != palette.REQUIRED_UNDONE:
status_item.setForeground(QColor(style.color))
table.setItem(r, 1, status_item)
for c, col in enumerate(metric_cols, start=2):
cell = QTableWidgetItem(self._slit_cell(row, col))
cell.setTextAlignment(Qt.AlignCenter)
table.setItem(r, c, cell)
# Keep the per-slit table from collapsing for many slits.
table.setMinimumHeight(160)
self._detail.addWidget(table)
[docs]
def _slit_cell(self, row, col):
"""
Format one per-slit metric cell.
Generated by JXP and Claude.
Args:
row (:obj:`dict`): The slit_table row.
col (:obj:`str`): The column key (e.g. ``rms`` or
``pixelflat mean``).
Returns:
str: The cell text (em-dash when missing).
"""
if col in ('center', 'slitord_id', 'rms'):
value = row.get(col)
return self._fmt(value)
# flats: "<correction> mean" / "<correction> rms"
corr, _, field = col.rpartition(' ')
metric = row.get('corrections', {}).get(corr, {})
return self._fmt(metric.get(field))
[docs]
@staticmethod
def _fmt(value):
"""
Format a numeric metric for display.
Generated by JXP and Claude.
Args:
value: The metric value (or ``None``).
Returns:
str: A short string, or an em-dash for missing values.
"""
if value is None:
return '—'
if isinstance(value, float):
return f'{value:.4g}'
return str(value)
[docs]
def _launch(self, argv, description, hint='viewer window'):
"""
Launch a command via the launcher (no-op if no launcher / command).
Generated by JXP and Claude.
Args:
argv (:obj:`list`): The command argv (or ``None``).
description (:obj:`str`): Human description for the activity bar.
hint (:obj:`str`, optional): Where the result appears (e.g.
``'Ginga window'``).
Returns:
None.
"""
if argv and self._launcher is not None:
self._launcher.launch(argv, description=description, hint=hint)