autocal: 新增单位与换算页(默认锁定)及轴页报警与事件日志

- 设备设置新增“单位与换算”页:显示各轴 UNITS、读回值、脉冲/mm 换算与核对状态;
  默认锁定,点击“修改”后才可编辑,应用前弹窗确认,旋转轴偏离标定值时额外警告
- 轴调试页:脉冲当量改为只读并提供入口;新增报警与事件日志(报警、使能、
  回零、读取失败、单位不符等)及单位核对摘要
- 新增 motion/unit_report.py、motion/events.py 及对应测试
This commit is contained in:
ykcheungab
2026-10-19 10:14:55 +08:00
parent c6c16df753
commit 22658dbfc1
12 changed files with 1462 additions and 10 deletions
@@ -0,0 +1,205 @@
"""Edges worth remembering from the controller's poll stream (Qt-free).
The axis table shows *now*. An alarm that came and went between two glances, a
drive that stopped answering for a second, or a homing run that ended without
completing leaves nothing on screen, and the operator is left with a machine that
"did something". The tracker turns the poll stream into a short list of dated
events: only changes, never one line per poll.
Nothing here talks to the controller; it reads :class:`~.models.Snapshot` values.
"""
from __future__ import annotations
import math
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from .models import AxisState, Snapshot
#: Event kinds. The page maps each one to a locale string.
CONNECTED = "connected"
DISCONNECTED = "disconnected"
ALARM_ON = "alarm_on"
ALARM_OFF = "alarm_off"
ENABLED = "enabled"
DISABLED = "disabled"
READ_FAILED = "read_failed"
READ_OK = "read_ok"
HOME_STARTED = "home_started"
HOME_DONE = "home_done"
HOME_INCOMPLETE = "home_incomplete"
UNITS_MISMATCH = "units_mismatch"
UNITS_OK = "units_ok"
#: How serious each kind is, for the colour the page gives the line.
SEVERITY = {
CONNECTED: "info",
DISCONNECTED: "warning",
ALARM_ON: "error",
ALARM_OFF: "info",
ENABLED: "info",
DISABLED: "info",
READ_FAILED: "warning",
READ_OK: "info",
HOME_STARTED: "info",
HOME_DONE: "info",
HOME_INCOMPLETE: "warning",
UNITS_MISMATCH: "error",
UNITS_OK: "info",
}
#: Platform-wide events carry this instead of an axis number.
NO_AXIS = -1
@dataclass(frozen=True)
class AxisEvent:
"""One dated change. ``detail`` is raw data (a reason, a number), not prose."""
time: float
axis: int
kind: str
detail: str = ""
@property
def severity(self) -> str:
return SEVERITY.get(self.kind, "info")
@dataclass
class _AxisMemory:
alarm: bool
enabled: bool
read_ok: bool
homing: bool
units_ok: bool | None
@dataclass
class AxisEventTracker:
"""Remember the last state of each axis and report what changed."""
limit: int = 200
clock: Callable[[], float] = time.time
_events: list[AxisEvent] = field(default_factory=list)
_connected: bool | None = None
_axes: dict[int, _AxisMemory] = field(default_factory=dict)
@property
def events(self) -> list[AxisEvent]:
"""Oldest first."""
return list(self._events)
def clear(self) -> None:
self._events.clear()
def update(
self,
snapshot: Snapshot,
expected_units: Mapping[int, float] | None = None,
) -> list[AxisEvent]:
"""Compare one poll with the previous one; return and keep the new events."""
fresh: list[AxisEvent] = []
now = float(self.clock())
def emit(axis: int, kind: str, detail: str = "") -> None:
fresh.append(AxisEvent(now, axis, kind, detail))
if self._connected is not None and snapshot.connected != self._connected:
emit(NO_AXIS, CONNECTED if snapshot.connected else DISCONNECTED)
self._connected = snapshot.connected
if not snapshot.connected:
# Whatever the axes did while the link was down is unknown: forget them,
# so the first poll after a reconnect is a baseline and not a flood.
self._axes.clear()
else:
for state in snapshot.axes:
self._update_axis(state, expected_units, emit)
self._events.extend(fresh)
if len(self._events) > self.limit:
del self._events[: len(self._events) - self.limit]
return fresh
def _update_axis(self, state: AxisState, expected_units, emit) -> None:
axis = state.axis
memory = self._axes.get(axis)
units_ok = self._units_ok(state, expected_units)
current = _AxisMemory(
alarm=bool(state.read_ok and state.status.alarm),
enabled=bool(state.read_ok and state.enabled),
read_ok=bool(state.read_ok),
homing=bool(state.read_ok and state.status.home_in_progress),
units_ok=units_ok,
)
if memory is None:
# The first sight of an axis is a baseline, except for the two facts that
# matter whenever they are true: an alarm already raised, and a drive that
# cannot be read.
if current.alarm:
emit(axis, ALARM_ON)
if not current.read_ok:
emit(axis, READ_FAILED, state.error)
if units_ok is False:
emit(axis, UNITS_MISMATCH, _units_detail(state, expected_units))
self._axes[axis] = current
return
if current.read_ok != memory.read_ok:
if current.read_ok:
emit(axis, READ_OK)
else:
emit(axis, READ_FAILED, state.error)
if current.read_ok and memory.read_ok:
if current.alarm != memory.alarm:
emit(axis, ALARM_ON if current.alarm else ALARM_OFF)
if current.enabled != memory.enabled:
emit(axis, ENABLED if current.enabled else DISABLED)
if current.homing != memory.homing:
if current.homing:
emit(axis, HOME_STARTED)
else:
emit(axis, HOME_DONE if state.home_status == 1 else HOME_INCOMPLETE)
if units_ok is not None and units_ok != memory.units_ok:
if units_ok:
if memory.units_ok is False:
emit(axis, UNITS_OK)
else:
emit(axis, UNITS_MISMATCH, _units_detail(state, expected_units))
self._axes[axis] = current
@staticmethod
def _units_ok(state: AxisState, expected_units) -> bool | None:
"""``None`` when there is nothing to compare, else whether the two agree."""
if expected_units is None or not state.read_ok:
return None
expected = expected_units.get(state.axis)
if expected is None:
return None
try:
expected = float(expected)
except (TypeError, ValueError):
return None
if not math.isfinite(expected) or expected <= 0:
return None
observed = float(state.units)
return bool(
math.isfinite(observed)
and observed > 0
and math.isclose(expected, observed, rel_tol=1e-7, abs_tol=0.0)
)
def _units_detail(state: AxisState, expected_units) -> str:
expected = (expected_units or {}).get(state.axis)
return f"{_g(expected)}|{_g(state.units)}"
def _g(value: object) -> str:
try:
return f"{float(value):g}"
except (TypeError, ValueError):
return "—"
@@ -0,0 +1,128 @@
"""What the platform's unit settings say, one row per direction (Qt-free).
The motion guards already refuse a move when the configured pulse equivalent
(``UNITS``) and the one the controller reports disagree. This module answers the
operator's question *before* a move is refused: which direction uses which drive,
what ``UNITS`` is configured, what the controller reads back, what one user unit
means in millimetres or degrees, and whether all of it holds together.
The verdicts follow :meth:`MotionService.units_problem` in the same order, so the
page and the guard cannot disagree about what is wrong. Nothing here writes to a
controller or changes a setting.
"""
from __future__ import annotations
import math
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from .models import Snapshot
from .service import AXIS_ROLES, LINEAR_ROLES
from .units import PULSES_PER_MM, units_per_degree
#: Relative tolerance of the expected-versus-read-back comparison. The guard uses
#: the same number; a different one here would show "ok" for a move it refuses.
UNITS_TOLERANCE = 1e-7
OK = "ok"
UNREAD = "unread"
MISMATCH = "mismatch"
UNCALIBRATED = "uncalibrated"
INVALID = "invalid"
UNMAPPED = "unmapped"
@dataclass(frozen=True)
class UnitRow:
"""One direction's unit facts.
``per_unit`` is millimetres (linear) or degrees (rotary) per controller user
unit, ``per_mm`` the inverse (user units per millimetre or per degree). Both
are NaN when the setting does not define them.
"""
role: str
axis: int
kind: str
expected: float
observed: float | None
status: str
per_unit: float
per_mm: float
@property
def ok(self) -> bool:
return self.status == OK
def build_unit_rows(
roles: Mapping[str, int],
expected_of: Callable[[int], float],
snapshot: Snapshot | None,
) -> list[UnitRow]:
"""One row per platform direction, in the order the machine is described."""
rows: list[UnitRow] = []
for role in AXIS_ROLES:
kind = "linear" if role in LINEAR_ROLES else "rotary"
axis = roles.get(role)
if axis is None:
rows.append(UnitRow(role, -1, kind, math.nan, None, UNMAPPED, math.nan, math.nan))
continue
axis = int(axis)
expected = _number(expected_of(axis))
observed = _observed(snapshot, axis)
per_unit, per_mm = _scale(kind, axis, expected)
status = _status(kind, axis, expected, observed)
rows.append(UnitRow(role, axis, kind, expected, observed, status, per_unit, per_mm))
return rows
def summarise(rows: list[UnitRow]) -> tuple[int, int, list[UnitRow]]:
"""``(rows that are fine, rows in all, the rows that are not)``."""
problems = [row for row in rows if not row.ok]
return len(rows) - len(problems), len(rows), problems
def _number(value: object) -> float:
try:
return float(value)
except (TypeError, ValueError):
return math.nan
def _observed(snapshot: Snapshot | None, axis: int) -> float | None:
if snapshot is None or not snapshot.connected:
return None
state = snapshot.axis(axis)
if state is None or not state.read_ok:
return None
return float(state.units)
def _scale(kind: str, axis: int, expected: float) -> tuple[float, float]:
"""Per-unit and per-millimetre (or per-degree) factors for the configured value."""
if not math.isfinite(expected) or expected <= 0:
return math.nan, math.nan
if kind == "linear":
return expected / PULSES_PER_MM, PULSES_PER_MM / expected
per_degree = units_per_degree(axis, units=expected)
if not math.isfinite(per_degree) or per_degree <= 0:
return math.nan, math.nan
return 1.0 / per_degree, per_degree
def _status(kind: str, axis: int, expected: float, observed: float | None) -> str:
if not math.isfinite(expected) or expected <= 0:
return INVALID
if kind == "rotary" and not math.isfinite(units_per_degree(axis, units=expected)):
return UNCALIBRATED
if observed is None:
return UNREAD
if (
not math.isfinite(observed)
or observed <= 0
or not math.isclose(expected, observed, rel_tol=UNITS_TOLERANCE, abs_tol=0.0)
):
return MISMATCH
return OK
@@ -15,6 +15,9 @@ five-axis platform, acting on the wrong axis is the expensive mistake.
from __future__ import annotations
import math
import time
from collections.abc import Iterable, Mapping
from dataclasses import replace
from orisys_app_shared.i18n import t
from PySide6.QtCore import Qt, Signal
@@ -25,6 +28,8 @@ from PySide6.QtWidgets import (
QHBoxLayout,
QLabel,
QLineEdit,
QListWidget,
QListWidgetItem,
QPushButton,
QTableWidget,
QTableWidgetItem,
@@ -42,6 +47,8 @@ from ..motion import (
safe_float,
user_units_to_mm,
)
from ..motion.events import UNITS_MISMATCH, AxisEvent, AxisEventTracker
from ..motion.unit_report import UNREAD, UnitRow, summarise
#: Data columns in the state table, in display order.
#:
@@ -113,6 +120,8 @@ class AxisPanel(QGroupBox):
moveRelRequested = Signal(int, float)
homeRequested = Signal(int)
clearAlarmRequested = Signal(int)
#: The operator wants the unit settings (shown and changed in the device setup).
unitsRequested = Signal()
def __init__(
self,
@@ -133,6 +142,8 @@ class AxisPanel(QGroupBox):
self._selected_axis = 0
self._snapshot = Snapshot()
self._rows: list[dict] = []
self._tracker = AxisEventTracker()
self._unit_rows: list[UnitRow] = []
outer = QVBoxLayout(self)
outer.setContentsMargins(10, 8, 10, 10)
@@ -140,11 +151,12 @@ class AxisPanel(QGroupBox):
outer.addWidget(self._build_params())
outer.addWidget(self._build_target_row())
# Stretch factor 1: the table is the only part of this panel that may
# grow, so every spare pixel goes to a further axis row rather than to
# padding around the parameter form.
outer.addWidget(self._build_table(), 1)
# The table is exactly as tall as its axis rows: the spare height belongs to
# the event log below, which is where an operator looks after something went
# wrong, and not to blank rows nobody can use.
outer.addWidget(self._build_table())
outer.addWidget(self._build_actions())
outer.addWidget(self._build_events(), 1)
self.retranslate()
self._reread_params()
@@ -181,6 +193,10 @@ class AxisPanel(QGroupBox):
self.units_label = label("axes.units")
self.units_edit = edit(80)
# Units are shown here and changed in the device setup, where they are locked
# until the operator asks to modify them: a pulse equivalent typed into a form
# that is always live rescales every position the machine is told about.
self.units_edit.setReadOnly(True)
self.speed_label = label("axes.speed")
self.speed_edit = edit(80)
self.accel_label = label("axes.accel")
@@ -211,6 +227,12 @@ class AxisPanel(QGroupBox):
lambda: self.clearAlarmRequested.emit(self.selected_axis)
)
self.units_button = QPushButton(box)
self.units_button.setObjectName("axisUnitsOpen")
self.units_button.setProperty("rail", True)
self.units_button.setFixedHeight(FIELD_HEIGHT)
self.units_button.clicked.connect(self.unitsRequested)
grid.addWidget(self.axis_label, 0, 0)
grid.addWidget(self.axis_index_label, 0, 1)
grid.addWidget(self.units_label, 0, 2)
@@ -232,6 +254,7 @@ class AxisPanel(QGroupBox):
grid.addWidget(self.apply_button, 2, 0, 1, 2)
grid.addWidget(self.home_button, 2, 2, 1, 2)
grid.addWidget(self.clear_alarm_button, 2, 4, 1, 2)
grid.addWidget(self.units_button, 2, 6, 1, 2)
return box
def _build_target_row(self) -> QWidget:
@@ -275,6 +298,7 @@ class AxisPanel(QGroupBox):
# and the splitter then clips the last drive — which reads as a missing
# axis. The table scrolls if it is ever squeezed below this.
self.table.setMinimumHeight(40 + ROW_HEIGHT * MIN_VISIBLE_ROWS)
self.table.setMaximumHeight(40 + ROW_HEIGHT * max(self.axis_count, MIN_VISIBLE_ROWS) + 4)
self.table.cellClicked.connect(self._on_cell_clicked)
for row in range(self.axis_count):
@@ -347,6 +371,97 @@ class AxisPanel(QGroupBox):
column.addLayout(row)
return box
def _build_events(self) -> QWidget:
box = QWidget(self)
column = QVBoxLayout(box)
column.setContentsMargins(0, 0, 0, 0)
column.setSpacing(4)
header = QHBoxLayout()
header.setSpacing(8)
self.events_title = QLabel(box)
self.events_title.setObjectName("statusValue")
self.units_chip = QLabel(box)
self.units_chip.setObjectName("hintLabel")
self.units_chip.setWordWrap(True)
self.clear_events_button = QPushButton(box)
self.clear_events_button.setObjectName("axisEventsClear")
self.clear_events_button.setProperty("rail", True)
self.clear_events_button.setFixedHeight(FIELD_HEIGHT)
self.clear_events_button.clicked.connect(self._on_clear_events)
header.addWidget(self.events_title)
header.addWidget(self.units_chip, 1)
header.addWidget(self.clear_events_button)
column.addLayout(header)
self.event_log = QListWidget(box)
self.event_log.setObjectName("axisEventLog")
self.event_log.setSelectionMode(QListWidget.SelectionMode.NoSelection)
self.event_log.setAlternatingRowColors(True)
self.event_log.setMinimumHeight(90)
column.addWidget(self.event_log, 1)
self.events_empty = QLabel(box)
self.events_empty.setObjectName("hintLabel")
self.events_empty.setWordWrap(True)
column.addWidget(self.events_empty)
return box
# ------------------------------------------------------------------ events
def _append_event(self, event: AxisEvent) -> None:
stamp = time.strftime("%H:%M:%S", time.localtime(event.time))
item = QListWidgetItem(f"{stamp} {event_text(event)}")
colour = _EVENT_COLOURS.get(event.severity)
if colour:
item.setForeground(QBrush(QColor(colour)))
self.event_log.insertItem(0, item)
while self.event_log.count() > self._tracker.limit:
self.event_log.takeItem(self.event_log.count() - 1)
def _render_events(self) -> None:
"""Rebuild the list from the tracker (language change, clear)."""
self.event_log.clear()
for event in self._tracker.events:
self._append_event(event)
self.events_empty.setVisible(self.event_log.count() == 0)
def _on_clear_events(self) -> None:
self._tracker.clear()
self._render_events()
def set_unit_rows(self, rows: Iterable[UnitRow], connected: bool) -> None:
"""One line about the unit check; the full table lives in the device setup."""
self._unit_rows = list(rows)
if not self._unit_rows:
self.units_chip.setText("")
return
_good, total, problems = summarise(self._unit_rows)
if not problems:
self.units_chip.setText(t("units.summary_ok", total=total))
elif all(row.status == UNREAD for row in problems):
self.units_chip.setText(t("units.summary_unread"))
else:
self.units_chip.setText(
t("units.summary_bad", bad=len(problems), total=total)
)
def apply_units(self, changes: Mapping[int, float]) -> None:
"""Send confirmed unit changes through the ordinary axis-parameter path."""
for axis, value in sorted(changes.items()):
if not 0 <= int(axis) < self.axis_count:
continue
params = replace(self._params.get(int(axis), AxisParams()), units=float(value))
self._params[int(axis)] = params
self.paramsApplied.emit(int(axis), params)
self._reread_params()
def expected_units(self, axis: int) -> float:
"""The pulse equivalent configured for a drive."""
return self._params.get(int(axis), AxisParams()).units
@property
def snapshot(self) -> Snapshot:
"""The last poll the panel rendered."""
return self._snapshot
# ------------------------------------------------------------------- state
@property
@@ -473,6 +588,10 @@ class AxisPanel(QGroupBox):
)
self._refresh_availability()
self._recolour_rows()
expected = {axis: self.expected_units(axis) for axis in range(self.axis_count)}
for event in self._tracker.update(snapshot, expected):
self._append_event(event)
self.events_empty.setVisible(self.event_log.count() == 0)
def _detail_text(self) -> str:
"""Everything about the selected axis that the seven columns cannot hold.
@@ -571,6 +690,12 @@ class AxisPanel(QGroupBox):
self.move_rel_button.setText(t("axes.move_rel"))
self.stop_all_button.setText(t("axes.stop_all"))
self.mm_hint.setText(t("axes.mm_derived_tip"))
self.units_button.setText(t("axes.units_open"))
self.units_edit.setToolTip(t("axes.units_locked_tip"))
self.events_title.setText(t("axes.events_title"))
self.clear_events_button.setText(t("axes.events_clear"))
self.events_empty.setText(t("axes.events_empty"))
self._render_events()
if hasattr(self, "_detail_label"):
self._detail_label.setText(self._detail_text())
self.table.setHorizontalHeaderLabels(
@@ -584,6 +709,19 @@ class AxisPanel(QGroupBox):
self._rows[row]["stop_button"].setText(t("axes.stop"))
#: Line colours for the event log. Info lines keep the theme's own text colour.
_EVENT_COLOURS = {"error": "#c62828", "warning": "#b26a00"}
def event_text(event: AxisEvent) -> str:
"""One event as a sentence in the current language."""
fields: dict[str, object] = {"axis": event.axis, "detail": event.detail}
if event.kind == UNITS_MISMATCH:
expected, _, observed = event.detail.partition("|")
fields.update(expected=expected or "—", observed=observed or "—")
return t(f"axes.ev_{event.kind}", **fields)
def _number(value: float, places: int) -> str:
try:
return f"{float(value):.{places}f}"
@@ -43,6 +43,7 @@ from ..motion import ConnectionSpec
from ..motion.service import AXIS_ROLES, DEFAULT_AXIS_ROLES, PLATFORM_AXES
from .bus_panel import BusPanel
from .connection_panel import ConnectionPanel
from .units_page import UnitsPage
#: Tab identity -> locale key, in the order they are shown. The order is the work:
#: reach the machine, bring its bus up, then say what each drive moves and how the
@@ -52,6 +53,7 @@ TABS: tuple[tuple[str, str], ...] = (
("bus", "setup.tab_bus"),
("axes", "setup.tab_axes"),
("press", "setup.tab_press"),
("units", "setup.tab_units"),
)
#: Role -> locale key for the mapping rows.
@@ -311,6 +313,7 @@ class DeviceSetupDialog(QDialog):
self.bus_panel = BusPanel(bas_path, self.tabs)
self.roles_page = AxisRolesPage(roles, self.tabs)
self.press_page = PressAxisPage(press, self.tabs)
self.units_page = UnitsPage(self.tabs)
for key, locale_key in TABS:
self.tabs.addTab(self.panel(key), "")
column.addWidget(self.tabs, 1)
@@ -331,11 +334,13 @@ class DeviceSetupDialog(QDialog):
# ----------------------------------------------------------------- content
def panel(self, key: str) -> QWidget:
"""One of the four pages, by tab key."""
"""One of the five pages, by tab key."""
if key == "connection":
return self.connection_panel
if key == "bus":
return self.bus_panel
if key == "units":
return self.units_page
return self.roles_page if key == "axes" else self.press_page
def open_for(self, tab: str | None = None) -> None:
@@ -402,3 +407,4 @@ class DeviceSetupDialog(QDialog):
self.bus_panel.retranslate()
self.roles_page.retranslate()
self.press_page.retranslate()
self.units_page.retranslate()
@@ -0,0 +1,405 @@
"""The units page of the device setup: what the platform's scales are, and whether they hold.
Every millimetre and degree the app shows or commands is *derived* from a drive's
pulse equivalent (``UNITS``) and a measured constant. If the equivalent configured here
and the one the controller reports differ, the motion guards refuse the move; this page
puts the same facts in front of the operator one step earlier, so a wrong scale is seen
at start-up and not as a refused move.
It is **locked by default**. Changing ``UNITS`` rescales every position the machine is
told about, so the cells are read-only until the operator presses *Modify*, and a change
is written only after a confirmation that lists what will change. The page itself
writes nothing: it emits :attr:`UnitsPage.unitsApplied` and the window sends the values
through the same axis-parameter path the axis page uses.
"""
from __future__ import annotations
import math
from collections.abc import Iterable
from orisys_app_shared.i18n import t
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QBrush, QColor
from PySide6.QtWidgets import (
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QPushButton,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from ..motion.unit_report import (
INVALID,
MISMATCH,
OK,
UNCALIBRATED,
UNMAPPED,
UNREAD,
UnitRow,
summarise,
)
from ..motion.units import (
PULSES_PER_MM,
ROTARY_CALIBRATION_UNITS,
ROTARY_UNITS_PER_90_DEG,
)
#: Table columns, in display order.
COLUMNS = (
"units.col_role",
"units.col_axis",
"units.col_kind",
"units.col_configured",
"units.col_observed",
"units.col_scale",
"units.col_check",
)
CONFIGURED_COLUMN = 3
ROW_HEIGHT = 34
_ROLE_KEYS = {
"x": "setup.role_x",
"y": "setup.role_y",
"z": "setup.role_z",
"yaw": "setup.role_yaw",
"pitch": "setup.role_pitch",
}
_STATUS_KEYS = {
OK: "units.status_ok",
UNREAD: "units.status_unread",
MISMATCH: "units.status_mismatch",
UNCALIBRATED: "units.status_uncalibrated",
INVALID: "units.status_invalid",
UNMAPPED: "units.status_unmapped",
}
_TIP_KEYS = {
UNREAD: "units.tip_unread",
MISMATCH: "units.tip_mismatch",
UNCALIBRATED: "units.tip_uncalibrated",
INVALID: "units.tip_invalid",
UNMAPPED: "units.tip_unmapped",
}
_GOOD = "#2e7d32"
_BAD = "#c62828"
_QUIET = "#7a7a7a"
def _g(value: float | None, places: int = 6) -> str:
"""A number as the operator would write it, or a dash when there is none."""
if value is None:
return "—"
try:
number = float(value)
except (TypeError, ValueError):
return "—"
return f"{number:.{places}g}" if math.isfinite(number) else "—"
class UnitsPage(QWidget):
"""Read-only unit table with a guarded way to change it."""
#: ``{drive: new UNITS}`` for the drives the operator changed and confirmed.
unitsApplied = Signal(object)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._rows: list[UnitRow] = []
self._connected = False
self._locked = True
self._unlocked_axes: dict[str, int] = {}
self._edits: dict[str, QLineEdit] = {}
column = QVBoxLayout(self)
column.setContentsMargins(8, 8, 8, 8)
column.setSpacing(8)
self.hint_label = QLabel(self)
self.hint_label.setObjectName("hintLabel")
self.hint_label.setWordWrap(True)
column.addWidget(self.hint_label)
top = QHBoxLayout()
top.setSpacing(8)
self.summary_label = QLabel(self)
self.summary_label.setObjectName("statusValue")
self.summary_label.setWordWrap(True)
self.modify_button = QPushButton(self)
self.modify_button.setObjectName("unitsModify")
self.modify_button.setProperty("rail", True)
self.modify_button.clicked.connect(self._unlock)
self.apply_button = QPushButton(self)
self.apply_button.setObjectName("unitsApply")
self.apply_button.setProperty("rail", True)
self.apply_button.clicked.connect(self._on_apply)
self.cancel_button = QPushButton(self)
self.cancel_button.setObjectName("unitsCancel")
self.cancel_button.setProperty("rail", True)
self.cancel_button.clicked.connect(self._on_cancel)
top.addWidget(self.summary_label, 1)
top.addWidget(self.modify_button)
top.addWidget(self.apply_button)
top.addWidget(self.cancel_button)
column.addLayout(top)
self.table = QTableWidget(5, len(COLUMNS), self)
self.table.setObjectName("unitsTable")
self.table.verticalHeader().setVisible(False)
self.table.setSelectionMode(QTableWidget.SelectionMode.NoSelection)
self.table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self.table.setAlternatingRowColors(True)
self.table.verticalHeader().setDefaultSectionSize(ROW_HEIGHT)
self.table.horizontalHeader().setStretchLastSection(True)
for index, width in enumerate((110, 60, 90, 110, 100, 230)):
self.table.setColumnWidth(index, width)
self.table.setMinimumHeight(40 + ROW_HEIGHT * 5 + 4)
for row in range(5):
for col in range(len(COLUMNS)):
if col != CONFIGURED_COLUMN:
item = QTableWidgetItem("")
if col in (1, 3, 4):
item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
self.table.setItem(row, col, item)
column.addWidget(self.table)
self.constants_label = QLabel(self)
self.constants_label.setObjectName("hintLabel")
self.constants_label.setWordWrap(True)
column.addWidget(self.constants_label)
self.message_label = QLabel(self)
self.message_label.setObjectName("hintLabel")
self.message_label.setWordWrap(True)
column.addWidget(self.message_label)
column.addStretch(1)
self._apply_lock()
self.retranslate()
# ------------------------------------------------------------------ data
@property
def locked(self) -> bool:
return self._locked
def set_rows(self, rows: Iterable[UnitRow], connected: bool) -> None:
"""Show one poll's worth of unit facts.
While the page is unlocked the configured column belongs to the operator and
is not overwritten by a refresh; if the drive mapping changes underneath an
unlocked edit, the edit is abandoned instead of being applied to another drive.
"""
self._rows = list(rows)
self._connected = bool(connected)
if not self._locked and {r.role: r.axis for r in self._rows} != self._unlocked_axes:
self._lock()
self.message_label.setText(t("units.mapping_changed"))
for index, row in enumerate(self._rows[: self.table.rowCount()]):
self._render_row(index, row)
self._render_summary()
self._apply_lock()
def _render_row(self, index: int, row: UnitRow) -> None:
def put(column: int, text: str, *, colour: str | None = None, tip: str = "") -> None:
item = self.table.item(index, column)
if item is None:
return
item.setText(text)
item.setToolTip(tip)
if colour:
item.setForeground(QBrush(QColor(colour)))
else:
item.setData(Qt.ItemDataRole.ForegroundRole, None)
put(0, t(_ROLE_KEYS.get(row.role, row.role)))
put(1, "—" if row.axis < 0 else str(row.axis))
put(2, t("units.kind_linear" if row.kind == "linear" else "units.kind_rotary"))
put(4, _g(row.observed, 6))
put(5, self._scale_text(row))
colour = _GOOD if row.ok else (_QUIET if row.status in {UNREAD, UNMAPPED} else _BAD)
put(
6,
t(_STATUS_KEYS[row.status]),
colour=colour,
tip=t(_TIP_KEYS[row.status]) if row.status in _TIP_KEYS else "",
)
edit = self._edit_for(index, row)
if edit is not None and (self._locked or not edit.isModified()):
edit.setText(_g(row.expected, 9))
edit.setModified(False)
def _edit_for(self, index: int, row: UnitRow) -> QLineEdit | None:
"""The configured-UNITS cell of one direction, created on first use."""
if row.axis < 0:
if self._edits.pop(row.role, None) is not None:
self.table.removeCellWidget(index, CONFIGURED_COLUMN)
return None
edit = self._edits.get(row.role)
if edit is None:
edit = QLineEdit(self.table)
edit.setObjectName(f"unitsEdit_{row.role}")
edit.setAlignment(Qt.AlignmentFlag.AlignCenter)
edit.setFixedHeight(28)
self._edits[row.role] = edit
self.table.setCellWidget(index, CONFIGURED_COLUMN, edit)
edit.setReadOnly(self._locked)
return edit
@staticmethod
def _scale_text(row: UnitRow) -> str:
if not math.isfinite(row.per_unit) or not math.isfinite(row.per_mm):
return "—"
if row.kind == "linear":
return t("units.scale_linear", mm=_g(row.per_unit), per=_g(row.per_mm))
return t("units.scale_rotary", deg=_g(row.per_unit), per=_g(row.per_mm))
def _render_summary(self) -> None:
if not self._rows:
self.summary_label.setText("")
return
good, total, problems = summarise(self._rows)
only_unread = bool(problems) and all(row.status == UNREAD for row in problems)
if not problems:
self.summary_label.setText(t("units.summary_ok", total=total))
colour = _GOOD
elif only_unread:
self.summary_label.setText(t("units.summary_unread"))
colour = _QUIET
else:
self.summary_label.setText(
t("units.summary_bad", bad=len(problems), total=total)
)
colour = _BAD
self.summary_label.setStyleSheet(f"color: {colour};")
# ---------------------------------------------------------------- locking
def _apply_lock(self) -> None:
for edit in self._edits.values():
edit.setReadOnly(self._locked)
self.modify_button.setVisible(self._locked)
self.modify_button.setEnabled(self._connected)
self.modify_button.setToolTip("" if self._connected else t("units.modify_offline"))
self.apply_button.setVisible(not self._locked)
self.apply_button.setEnabled(self._connected)
self.cancel_button.setVisible(not self._locked)
def _unlock(self) -> None:
if not self._connected:
return
self._locked = False
self._unlocked_axes = {row.role: row.axis for row in self._rows}
self.message_label.setText(t("units.unlocked_note"))
self._apply_lock()
def _lock(self) -> None:
self._locked = True
self._unlocked_axes = {}
for row in self._rows:
edit = self._edits.get(row.role)
if edit is not None:
edit.setText(_g(row.expected, 9))
edit.setModified(False)
self._apply_lock()
def _on_cancel(self) -> None:
self._lock()
self.message_label.setText(t("units.locked_note"))
# ------------------------------------------------------------------ apply
def collect_changes(self) -> tuple[dict[int, float], str]:
"""``(changes, problem)`` from the typed values; a problem means nothing is applied."""
changes: dict[int, float] = {}
for row in self._rows:
edit = self._edits.get(row.role)
if edit is None or row.axis < 0:
continue
try:
value = float(edit.text().strip())
except ValueError:
return {}, t("units.invalid", role=t(_ROLE_KEYS.get(row.role, row.role)))
if not math.isfinite(value) or value <= 0:
return {}, t("units.invalid", role=t(_ROLE_KEYS.get(row.role, row.role)))
if not math.isclose(value, row.expected, rel_tol=1e-9, abs_tol=0.0):
changes[row.axis] = value
return changes, ""
def _on_apply(self) -> None:
changes, problem = self.collect_changes()
if problem:
self.message_label.setText(problem)
return
if not changes:
self._lock()
self.message_label.setText(t("units.no_change"))
return
if not self._confirm(changes):
return
self._lock()
self.message_label.setText(t("units.applied", count=len(changes)))
self.unitsApplied.emit(dict(changes))
def _confirm(self, changes: dict[int, float]) -> bool:
by_axis = {row.axis: row for row in self._rows if row.axis >= 0}
lines = []
off_calibration = []
for axis, value in sorted(changes.items()):
row = by_axis.get(axis)
role = t(_ROLE_KEYS.get(row.role, row.role)) if row else "?"
lines.append(
t("units.change_line", role=role, axis=axis, old=_g(row.expected if row else None, 9), new=_g(value, 9))
)
if (
row is not None
and row.kind == "rotary"
and axis in ROTARY_UNITS_PER_90_DEG
and not math.isclose(value, ROTARY_CALIBRATION_UNITS, rel_tol=1e-7, abs_tol=0.0)
):
off_calibration.append(role)
body = t("units.confirm_body", changes="\n".join(lines))
if off_calibration:
body += "\n\n" + t(
"units.confirm_rotary",
roles=", ".join(off_calibration),
cal=_g(ROTARY_CALIBRATION_UNITS),
)
answer = QMessageBox.question(
self,
t("units.confirm_title"),
body,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
return answer == QMessageBox.StandardButton.Yes
# ----------------------------------------------------------- translations
def retranslate(self) -> None:
self.hint_label.setText(t("units.hint"))
self.modify_button.setText(t("units.modify"))
self.apply_button.setText(t("units.apply"))
self.cancel_button.setText(t("units.cancel"))
self.table.setHorizontalHeaderLabels([t(key) for key in COLUMNS])
rotary = ";".join(
t("units.rotary_item", axis=axis, units=_g(value, 9))
for axis, value in sorted(ROTARY_UNITS_PER_90_DEG.items())
)
self.constants_label.setText(
t(
"units.constants",
ppm=_g(PULSES_PER_MM, 9),
rotary=rotary,
cal=_g(ROTARY_CALIBRATION_UNITS),
)
)
if not self.message_label.text():
self.message_label.setText(t("units.locked_note"))
for index, row in enumerate(self._rows[: self.table.rowCount()]):
self._render_row(index, row)
self._render_summary()
self._apply_lock()
@@ -46,6 +46,7 @@ from ..motion.pose_move import (
press_targets,
)
from ..motion.service import clamp_bas_task
from ..motion.unit_report import build_unit_rows
from ..motion.units import DEFAULT_UNITS, user_units_to_deg, user_units_to_mm
from ..preferences import load_preferences, save_preferences
from ..sensor import (
@@ -367,6 +368,8 @@ class AutoCalMainWindow(QMainWindow):
self.axis_panel.moveRelRequested.connect(self._on_move_rel)
self.axis_panel.homeRequested.connect(self._on_home)
self.axis_panel.clearAlarmRequested.connect(self._on_clear_alarm)
self.axis_panel.unitsRequested.connect(lambda: self._show_device_setup("units"))
self.device_setup.units_page.unitsApplied.connect(self._on_units_applied)
self.bus_data_panel.sdoReadRequested.connect(self._on_sdo_read)
self.bus_data_panel.sdoWriteRequested.connect(self._on_sdo_write)
self.bus_data_panel.rtexReadRequested.connect(self._on_rtex_read)
@@ -423,6 +426,28 @@ class AutoCalMainWindow(QMainWindow):
problem = self.poller.service.roles_problem()
if problem:
self.finger_panel.set_move_status(f"✗ {problem}")
self._refresh_unit_report()
def _refresh_unit_report(self) -> None:
"""Feed the unit table and the axis page's one-line check from one computation.
Both read the same rows, so the line on the axis page and the table in the
device setup cannot disagree. Nothing here writes to the controller.
"""
panel = getattr(self, "axis_panel", None)
if panel is None:
return
snapshot = panel.snapshot
rows = build_unit_rows(
self.device_setup.roles_page.roles(), panel.expected_units, snapshot
)
self.device_setup.units_page.set_rows(rows, snapshot.connected)
panel.set_unit_rows(rows, snapshot.connected)
def _on_units_applied(self, changes: dict) -> None:
"""Confirmed unit changes go out through the ordinary axis-parameter path."""
self.axis_panel.apply_units(changes)
self._refresh_unit_report()
def _capture_bus_problem(self) -> str:
"""Why the capture page's bus check fails, or ``""`` when it passes.
@@ -942,6 +967,7 @@ class AutoCalMainWindow(QMainWindow):
self.bus_data_panel.set_connected(snapshot.connected)
self.bus_data_panel.set_bus_type(snapshot.bus.bus_type)
self.axis_panel.update_snapshot(snapshot)
self._refresh_unit_report()
self.axis_readout.update_snapshot(snapshot)
# Progress the pose move. It waits on the axis readings in this same
# snapshot, so it costs the poll nothing extra and cannot run ahead of the