feat(da07): channels grid adopts status tags, alarm rows, active toggle, units, summary, write feedback
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -17,6 +17,7 @@ from PySide6.QtWidgets import QComboBox, QHBoxLayout, QLabel, QPushButton, QVBox
|
||||
from cim_suite.core.export import Cell, Sheet
|
||||
from cim_suite.core.sensor_models import identify, layout
|
||||
from cim_suite.core.ui.group_band_delegate import GroupBandDelegate
|
||||
from cim_suite.core.ui.kit import ALARM_ROW_ROLE, STATUS_ROLE, CellKind
|
||||
from cim_suite.core.ui.table_tab import TableTab
|
||||
from cim_suite.core.ui.theme import current, qcolor
|
||||
|
||||
@@ -81,7 +82,7 @@ def staleness(elapsed_s: float) -> str:
|
||||
|
||||
class ChannelsTab(TableTab):
|
||||
def __init__(self, controller, *, repository=None) -> None:
|
||||
super().__init__(HEADERS, editable_cols=_EDIT_MAP.keys())
|
||||
super().__init__(HEADERS, editable_cols=_EDIT_MAP.keys(), checkable_cols=(1,))
|
||||
self._ctrl = controller
|
||||
self._repo = repository
|
||||
self._device: int | None = None
|
||||
@@ -89,6 +90,19 @@ class ChannelsTab(TableTab):
|
||||
self._delegate = GroupBandDelegate(self.table)
|
||||
self.table.setItemDelegate(self._delegate)
|
||||
self.delegate = self._delegate # keep TableTab.delegate pointing at the installed one
|
||||
|
||||
d = self._delegate
|
||||
d.set_column_kind(0, CellKind.NUMERIC)
|
||||
d.set_column_kind(1, CellKind.TOGGLE) # checkable col, on the REAL delegate
|
||||
d.set_column_kind(3, CellKind.IDENTIFIER) # Serial — recessive mono
|
||||
for col in (5, 6, 7, 9, 10, 11, 12, 13, 14, 15, 16, 17):
|
||||
d.set_column_kind(col, CellKind.NUMERIC)
|
||||
d.set_column_kind(_STATUS_COL, CellKind.STATUS) # §1.2 tag
|
||||
|
||||
self.set_column_units({"Scale": "×", "Offset": "+", "Input": "raw", "Refresh": "s"})
|
||||
self.enable_summary("✎ Click a value to edit · Enter saves · Esc cancels")
|
||||
self._pending_writes: set[tuple[int, int]] = set()
|
||||
|
||||
from cim_suite.core.ui.copy_menu import set_extra_actions
|
||||
|
||||
set_extra_actions(self.table, self._row_actions)
|
||||
@@ -179,17 +193,31 @@ class ChannelsTab(TableTab):
|
||||
return ordered, rows, grouped_rows, starts
|
||||
|
||||
def rebuild(self) -> None:
|
||||
for r, c in self._pending_writes: # §5.6: the echo IS the confirmation
|
||||
self.delegate.resolve(r, c, ok=True)
|
||||
self._pending_writes.clear()
|
||||
|
||||
if self._device is None:
|
||||
self._channels = []
|
||||
self.set_rows([])
|
||||
self._delegate.set_groups([], [])
|
||||
self.summary.set_summary([("0 channels", None)])
|
||||
return
|
||||
self._channels, rows, grouped_rows, starts = self._rows_for_device(self._device)
|
||||
self.set_rows(rows)
|
||||
self._delegate.set_groups(grouped_rows, starts)
|
||||
self._colorize()
|
||||
self._apply_status_roles()
|
||||
self._tick_refresh()
|
||||
|
||||
warn = sum(1 for ch in self._channels if status_severity(ch.status) == "HW")
|
||||
alarm = sum(1 for ch in self._channels if status_severity(ch.status) == "ER")
|
||||
segments: list[tuple[str, str | None]] = [(f"{len(self._channels)} channels", None)]
|
||||
if warn:
|
||||
segments.append((f"{warn} warn", "warn"))
|
||||
if alarm:
|
||||
segments.append((f"{alarm} alarm", "alarm"))
|
||||
self.summary.set_summary(segments)
|
||||
|
||||
def export_sheets(self) -> list[Sheet]:
|
||||
"""One worksheet per present device — so Export-All covers every configured
|
||||
device regardless of which one is currently shown."""
|
||||
@@ -205,19 +233,22 @@ class ChannelsTab(TableTab):
|
||||
))
|
||||
return sheets
|
||||
|
||||
def _colorize(self) -> None:
|
||||
self._loading = True # suppress itemChanged while we tint cells
|
||||
def _apply_status_roles(self) -> None:
|
||||
"""§1.2: status tag on the Status cell; full-row tint for alarm-class only
|
||||
(severity via the verified BL-E8 mapping — fault/ALARM red, pure WARN amber)."""
|
||||
self._loading = True
|
||||
try:
|
||||
for r, ch in enumerate(self._channels):
|
||||
sev = status_severity(ch.status)
|
||||
if sev is None:
|
||||
continue
|
||||
it = self.table.item(r, _STATUS_COL)
|
||||
if it:
|
||||
fg_t, fill_t = current().severity["HW" if sev == "HW" else "ER"]
|
||||
fg, fill = qcolor(fg_t), qcolor(fill_t)
|
||||
it.setBackground(fill)
|
||||
it.setForeground(fg)
|
||||
status = {"ER": "alarm", "HW": "warn"}.get(sev, "ok")
|
||||
cell = self.table.item(r, _STATUS_COL)
|
||||
if cell is not None:
|
||||
cell.setData(STATUS_ROLE, status)
|
||||
if not cell.text():
|
||||
cell.setText("OK")
|
||||
first = self.table.item(r, 0)
|
||||
if first is not None:
|
||||
first.setData(ALARM_ROW_ROLE, status == "alarm")
|
||||
finally:
|
||||
self._loading = False
|
||||
|
||||
@@ -245,13 +276,19 @@ class ChannelsTab(TableTab):
|
||||
def on_edit(self, row: int, col: int, value: str) -> None:
|
||||
if self._device is None or row >= len(self._channels):
|
||||
return
|
||||
channel = self._channels[row].channel
|
||||
if col == 1: # Active toggle
|
||||
self._ctrl.set_channel_active(self._device, channel, value.strip() in ("1", "true", "True"))
|
||||
return
|
||||
setter = _EDIT_MAP.get(col)
|
||||
if setter is not None and col != 1:
|
||||
getattr(self._ctrl, setter)(self._device, channel, value)
|
||||
if setter is None:
|
||||
return
|
||||
getattr(self._ctrl, setter)(self._device, self._channels[row].channel, value)
|
||||
self.delegate.mark_pending(row, col)
|
||||
self._pending_writes.add((row, col))
|
||||
|
||||
def on_check(self, row: int, col: int, checked: bool) -> None:
|
||||
if col != 1 or self._device is None or row >= len(self._channels):
|
||||
return
|
||||
self._ctrl.set_channel_active(self._device, self._channels[row].channel, checked)
|
||||
self.delegate.mark_pending(row, col)
|
||||
self._pending_writes.add((row, col))
|
||||
|
||||
def _calibrate(self) -> None:
|
||||
row = self.table.currentRow()
|
||||
|
||||
85
tests/da07/test_channels_tab_status.py
Normal file
85
tests/da07/test_channels_tab_status.py
Normal file
@@ -0,0 +1,85 @@
|
||||
"""Channels tab: §1.2 status tags / alarm rows, §5.4 units + summary, §5.6 feedback."""
|
||||
|
||||
from PySide6.QtCore import Qt
|
||||
from PySide6.QtWidgets import QLabel
|
||||
|
||||
from cim_suite.core.ui.kit import ALARM_ROW_ROLE, STATUS_ROLE
|
||||
from cim_suite.modules.da07.domain.controller import StationController
|
||||
from cim_suite.modules.da07.transport.simulator import SimulatedStation
|
||||
from cim_suite.modules.da07.ui.channels_tab import ChannelsTab
|
||||
|
||||
|
||||
def _tab(qtbot, devices=1, channels=3):
|
||||
sim = SimulatedStation(devices=devices, channels=channels)
|
||||
ctrl = StationController()
|
||||
ctrl.attach(sim)
|
||||
sim.start()
|
||||
ctrl.refresh()
|
||||
tab = ChannelsTab(ctrl)
|
||||
qtbot.addWidget(tab)
|
||||
return tab, ctrl
|
||||
|
||||
|
||||
def _set_statuses(tab, ctrl, statuses):
|
||||
for i, status in enumerate(statuses):
|
||||
ctrl.channels.get(0, i).status = status
|
||||
tab.rebuild()
|
||||
|
||||
|
||||
def test_status_cell_carries_status_role(qtbot):
|
||||
tab, ctrl = _tab(qtbot)
|
||||
_set_statuses(tab, ctrl, ["OK", "WARN", "ALARM"])
|
||||
assert tab.table.item(0, 8).data(STATUS_ROLE) == "ok"
|
||||
assert tab.table.item(1, 8).data(STATUS_ROLE) == "warn"
|
||||
assert tab.table.item(2, 8).data(STATUS_ROLE) == "alarm"
|
||||
|
||||
|
||||
def test_alarm_rows_flag_full_row_tint_warn_rows_do_not(qtbot):
|
||||
tab, ctrl = _tab(qtbot)
|
||||
_set_statuses(tab, ctrl, ["OK", "WARN", "ALARM"])
|
||||
assert not tab.table.item(0, 0).data(ALARM_ROW_ROLE)
|
||||
assert not tab.table.item(1, 0).data(ALARM_ROW_ROLE)
|
||||
assert tab.table.item(2, 0).data(ALARM_ROW_ROLE)
|
||||
|
||||
|
||||
def test_sensor_fault_is_alarm_class(qtbot):
|
||||
tab, ctrl = _tab(qtbot, channels=1)
|
||||
_set_statuses(tab, ctrl, ["Open"])
|
||||
assert tab.table.item(0, 8).data(STATUS_ROLE) == "alarm"
|
||||
assert tab.table.item(0, 0).data(ALARM_ROW_ROLE)
|
||||
|
||||
|
||||
def test_active_column_is_a_toggle_and_writes(qtbot, monkeypatch):
|
||||
tab, ctrl = _tab(qtbot, channels=2)
|
||||
item = tab.table.item(0, 1)
|
||||
assert item.flags() & Qt.ItemFlag.ItemIsUserCheckable
|
||||
sent = []
|
||||
monkeypatch.setattr(
|
||||
ctrl, "set_channel_active", lambda dev, ch, on: sent.append((dev, ch, on))
|
||||
)
|
||||
item.setCheckState(Qt.CheckState.Unchecked)
|
||||
assert sent == [(0, 0, False)]
|
||||
|
||||
|
||||
def test_summary_counts(qtbot):
|
||||
tab, ctrl = _tab(qtbot)
|
||||
_set_statuses(tab, ctrl, ["OK", "WARN", "ALARM"])
|
||||
texts = {
|
||||
lbl.text() for lbl in tab.summary.findChildren(QLabel)
|
||||
if lbl.objectName() == "SummaryItem"
|
||||
}
|
||||
assert "3 CHANNELS" in texts and "1 WARN" in texts and "1 ALARM" in texts
|
||||
|
||||
|
||||
def test_edit_marks_pending_and_rebuild_resolves(qtbot):
|
||||
tab, _ = _tab(qtbot)
|
||||
tab.on_edit(0, 2, "Inlet") # Tag edit
|
||||
assert (0, 2) in tab.delegate._pending
|
||||
tab.rebuild()
|
||||
assert (0, 2) not in tab.delegate._pending
|
||||
|
||||
|
||||
def test_units_row_present(qtbot):
|
||||
tab, _ = _tab(qtbot)
|
||||
units = tab.table.horizontalHeader()._units
|
||||
assert units[5] == "×" and units[6] == "+" and units[17] == "s"
|
||||
Reference in New Issue
Block a user