diff --git a/apps/autocal/config/locales/en_US.json b/apps/autocal/config/locales/en_US.json index aac4235..3193ff4 100644 --- a/apps/autocal/config/locales/en_US.json +++ b/apps/autocal/config/locales/en_US.json @@ -51,7 +51,8 @@ "press_set": "Calibrated: press target = zero ± contact height.", "press_unset": "Not calibrated: Execute holds the press axis.", "close": "Close", - "chip_hint": "Open device setup — {name}" + "chip_hint": "Open device setup — {name}", + "tab_units": "Units" }, "move": { "execute": "Execute", @@ -164,7 +165,25 @@ "home": "Home", "controls": "Actions", "mm_derived_tip": "Millimetres are derived from the pulse equivalent: a display value, not a calibrated length.", - "enabled_count": "{on}/{total} axes enabled" + "enabled_count": "{on}/{total} axes enabled", + "units_open": "Units…", + "units_locked_tip": "Units are locked and read-only here. Use Device setup → Units and press Modify to change them.", + "events_title": "Alarms and events", + "events_clear": "Clear log", + "events_empty": "No events yet. Changes in alarms, enable, homing, read failures and unit mismatches are logged here.", + "ev_connected": "Controller connected", + "ev_disconnected": "Controller disconnected", + "ev_alarm_on": "Axis {axis} alarm", + "ev_alarm_off": "Axis {axis} alarm cleared", + "ev_enabled": "Axis {axis} enabled", + "ev_disabled": "Axis {axis} disabled", + "ev_read_failed": "Axis {axis} read failed: {detail}", + "ev_read_ok": "Axis {axis} readable again", + "ev_home_started": "Axis {axis} homing started", + "ev_home_done": "Axis {axis} homing complete", + "ev_home_incomplete": "Axis {axis} homing ended without reporting complete", + "ev_units_mismatch": "Axis {axis} UNITS mismatch: configured {expected}, read {observed}", + "ev_units_ok": "Axis {axis} UNITS agree again" }, "busdata": { "title": "Bus data exchange", @@ -911,5 +930,49 @@ "close": "Close", "max_required": "Required; the program does not assume a value" } + }, + "units": { + "hint": "One place to check each direction's drive, its pulse equivalent (UNITS) and the resulting scale. Locked and read-only by default; press Modify to change a value, and the change is listed for confirmation before anything is written.", + "col_role": "Direction", + "col_axis": "Drive", + "col_kind": "Type", + "col_configured": "Configured UNITS", + "col_observed": "Controller read-back", + "col_scale": "Scale", + "col_check": "Check", + "kind_linear": "Linear (mm)", + "kind_rotary": "Rotary (°)", + "scale_linear": "1 unit = {mm} mm; {per} units/mm", + "scale_rotary": "1 unit = {deg}°; {per} units/°", + "status_ok": "Agree", + "status_unread": "Not read", + "status_mismatch": "Mismatch", + "status_uncalibrated": "No angle scale", + "status_invalid": "Invalid", + "status_unmapped": "Unmapped", + "tip_unread": "Not connected, or this axis's UNITS has not been read back yet.", + "tip_mismatch": "The controller reads a different UNITS from the configured one: pose moves are refused and targets are not rescaled.", + "tip_uncalibrated": "The rotary angle scale was only measured at the calibration equivalent; any other value needs a new calibration.", + "tip_invalid": "UNITS must be a finite number above 0.", + "tip_unmapped": "No drive is assigned to this direction; see Axis mapping.", + "summary_ok": "Unit check: all {total} directions agree", + "summary_bad": "Unit check: {bad} of {total} directions need attention", + "summary_unread": "Unit check: not read yet (not connected)", + "constants": "Fixed calibration constants (measured on the bench, not editable here): linear axes {ppm} pulses/mm; rotary axes {rotary}; all measured at UNITS={cal}. Millimetres and degrees are displayed values derived from these constants, not new measurements.", + "rotary_item": "axis {axis}: {units} user units = 90°", + "modify": "Modify", + "apply": "Apply to controller", + "cancel": "Cancel", + "modify_offline": "Not connected to a controller; nothing can be modified.", + "locked_note": "Locked: read-only.", + "unlocked_note": "Unlocked: enter new values in the Configured UNITS column, then press Apply to controller.", + "mapping_changed": "The axis mapping changed; unapplied edits were dropped and the page is locked again.", + "invalid": "UNITS for {role} is not valid: enter a number above 0.", + "no_change": "No change; locked again.", + "applied": "Submitted UNITS for {count} drive(s); the read-back will show after the next refresh.", + "change_line": "{role} · axis {axis}: {old} → {new}", + "confirm_title": "Write UNITS", + "confirm_body": "The following UNITS will be written to the controller:\n{changes}\n\nAfterwards every millimetre and degree reading, and every pose-move target, is recomputed with the new equivalent. Confirm that the platform is stopped and the mechanical ratio has been checked.", + "confirm_rotary": "Note: UNITS for {roles} is not the calibration equivalent {cal}; the rotary axis has no angle scale at that value, so pose moves are refused until it is recalibrated." } } diff --git a/apps/autocal/config/locales/zh_CN.json b/apps/autocal/config/locales/zh_CN.json index 159158c..8f9c95d 100644 --- a/apps/autocal/config/locales/zh_CN.json +++ b/apps/autocal/config/locales/zh_CN.json @@ -51,7 +51,8 @@ "press_set": "已标定:压轴目标 = 零点 ± 接触高度。", "press_unset": "未标定:执行动作会保持压轴不动。", "close": "关闭", - "chip_hint": "点击打开设备设置 — {name}" + "chip_hint": "点击打开设备设置 — {name}", + "tab_units": "单位与换算" }, "move": { "execute": "执行动作", @@ -164,7 +165,25 @@ "home": "启动回零", "controls": "操作", "mm_derived_tip": "毫米值由脉冲当量换算得出,是显示值,不是标定长度。", - "enabled_count": "已使能 {on}/{total} 轴" + "enabled_count": "已使能 {on}/{total} 轴", + "units_open": "单位与换算…", + "units_locked_tip": "单位已锁定,这里只读。到“设备设置 → 单位与换算”点“修改”才能更改。", + "events_title": "报警与事件", + "events_clear": "清空记录", + "events_empty": "暂无事件。报警、使能、回零、读取失败和单位不一致发生变化时,会记录在这里。", + "ev_connected": "控制器已连接", + "ev_disconnected": "控制器已断开", + "ev_alarm_on": "轴 {axis} 报警", + "ev_alarm_off": "轴 {axis} 报警已解除", + "ev_enabled": "轴 {axis} 已使能", + "ev_disabled": "轴 {axis} 已关闭使能", + "ev_read_failed": "轴 {axis} 读取失败:{detail}", + "ev_read_ok": "轴 {axis} 读取恢复", + "ev_home_started": "轴 {axis} 开始回零", + "ev_home_done": "轴 {axis} 回零完成", + "ev_home_incomplete": "轴 {axis} 回零结束,但未报告完成", + "ev_units_mismatch": "轴 {axis} UNITS 不一致:配置 {expected},读回 {observed}", + "ev_units_ok": "轴 {axis} UNITS 已一致" }, "busdata": { "title": "总线数据交互", @@ -911,5 +930,49 @@ "close": "关闭", "max_required": "必须填写,程序不预设" } + }, + "units": { + "hint": "集中检查每个方向用的驱动器、脉冲当量(UNITS)和换算。默认锁定为只读;需要改动时点“修改”,写入前会列出改动并要求确认。", + "col_role": "方向", + "col_axis": "驱动器", + "col_kind": "类型", + "col_configured": "配置 UNITS", + "col_observed": "控制器读回", + "col_scale": "换算", + "col_check": "检查", + "kind_linear": "线性 (mm)", + "kind_rotary": "旋转 (°)", + "scale_linear": "1 单位 = {mm} mm;{per} 单位/mm", + "scale_rotary": "1 单位 = {deg}°;{per} 单位/°", + "status_ok": "一致", + "status_unread": "未读回", + "status_mismatch": "不一致", + "status_uncalibrated": "无角度标定", + "status_invalid": "配置无效", + "status_unmapped": "未映射", + "tip_unread": "未连接,或该轴的 UNITS 尚未成功读回。", + "tip_mismatch": "控制器读回的 UNITS 与配置不同:姿态移动会被拒绝,不会重新换算目标。", + "tip_uncalibrated": "旋转轴的角度比例只在标定当量下测量过;其他值需要重新标定。", + "tip_invalid": "UNITS 必须是大于 0 的有限数。", + "tip_unmapped": "该方向没有分配驱动器,见“轴映射”。", + "summary_ok": "单位检查:{total} 个方向全部一致", + "summary_bad": "单位检查:{bad}/{total} 个方向需要处理", + "summary_unread": "单位检查:尚未读回(未连接)", + "constants": "固定标定常数(来自实机测量,不能在此修改):线性轴 {ppm} 脉冲/mm;旋转轴 {rotary};以上在 UNITS={cal} 下测得。毫米与角度都是由这些常数换算的显示值,不是新的测量。", + "rotary_item": "轴 {axis}:{units} 用户单位 = 90°", + "modify": "修改", + "apply": "应用到控制器", + "cancel": "取消", + "modify_offline": "未连接控制器,无法修改。", + "locked_note": "已锁定:只读显示。", + "unlocked_note": "已解锁:在“配置 UNITS”列输入新值,然后点“应用到控制器”。", + "mapping_changed": "轴映射已改变,未应用的修改已放弃并重新锁定。", + "invalid": "{role} 的 UNITS 无效:需要大于 0 的数字。", + "no_change": "没有改动,已重新锁定。", + "applied": "已提交 {count} 个驱动器的 UNITS;读回结果将在下一次刷新后显示。", + "change_line": "{role} · 轴 {axis}:{old} → {new}", + "confirm_title": "写入 UNITS", + "confirm_body": "将向控制器写入以下 UNITS:\n{changes}\n\n写入后,全部毫米和角度读数以及姿态移动的目标都按新当量重新计算。请确认平台已停止,并已核对机械传动比。", + "confirm_rotary": "注意:{roles} 的 UNITS 不是标定当量 {cal},旋转轴在该当量下没有角度标定,姿态移动会被拒绝,直到重新标定。" } } diff --git a/apps/autocal/src/orisys_autocal/motion/events.py b/apps/autocal/src/orisys_autocal/motion/events.py new file mode 100644 index 0000000..f431a68 --- /dev/null +++ b/apps/autocal/src/orisys_autocal/motion/events.py @@ -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 "—" diff --git a/apps/autocal/src/orisys_autocal/motion/unit_report.py b/apps/autocal/src/orisys_autocal/motion/unit_report.py new file mode 100644 index 0000000..697e9ad --- /dev/null +++ b/apps/autocal/src/orisys_autocal/motion/unit_report.py @@ -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 diff --git a/apps/autocal/src/orisys_autocal/ui/axis_panel.py b/apps/autocal/src/orisys_autocal/ui/axis_panel.py index dfbd5ff..0638b5f 100644 --- a/apps/autocal/src/orisys_autocal/ui/axis_panel.py +++ b/apps/autocal/src/orisys_autocal/ui/axis_panel.py @@ -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}" diff --git a/apps/autocal/src/orisys_autocal/ui/device_setup.py b/apps/autocal/src/orisys_autocal/ui/device_setup.py index 34df33b..8c58e62 100644 --- a/apps/autocal/src/orisys_autocal/ui/device_setup.py +++ b/apps/autocal/src/orisys_autocal/ui/device_setup.py @@ -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() diff --git a/apps/autocal/src/orisys_autocal/ui/units_page.py b/apps/autocal/src/orisys_autocal/ui/units_page.py new file mode 100644 index 0000000..2e1b4a4 --- /dev/null +++ b/apps/autocal/src/orisys_autocal/ui/units_page.py @@ -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() diff --git a/apps/autocal/src/orisys_autocal/ui/window.py b/apps/autocal/src/orisys_autocal/ui/window.py index 9113777..6d1203a 100644 --- a/apps/autocal/src/orisys_autocal/ui/window.py +++ b/apps/autocal/src/orisys_autocal/ui/window.py @@ -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 diff --git a/apps/autocal/tests/test_axis_events.py b/apps/autocal/tests/test_axis_events.py new file mode 100644 index 0000000..d93a2cf --- /dev/null +++ b/apps/autocal/tests/test_axis_events.py @@ -0,0 +1,95 @@ +"""The axis event log: only changes, a baseline on first sight, nothing across a gap.""" + +from __future__ import annotations + +from orisys_autocal.motion import AxisState, Snapshot +from orisys_autocal.motion import events as ev +from orisys_autocal.motion.models import AxisStatusBits +from orisys_autocal.motion.events import AxisEventTracker + + +def _axis(axis=0, *, alarm=False, enabled=True, read_ok=True, homing=False, home_status=0, units=10.0, error=""): + raw = (512 if alarm else 0) | (64 if homing else 0) + return AxisState( + axis=axis, + enabled=enabled, + units=units, + status=AxisStatusBits(raw), + home_status=home_status, + read_ok=read_ok, + error=error, + ) + + +def _snap(*axes, connected=True) -> Snapshot: + return Snapshot(connected=connected, axes=tuple(axes)) + + +def _kinds(events): + return [(event.axis, event.kind) for event in events] + + +def test_the_first_poll_is_a_baseline_and_a_steady_state_is_silent(): + tracker = AxisEventTracker(clock=lambda: 1.0) + assert tracker.update(_snap(_axis(0), _axis(1))) == [] + assert tracker.update(_snap(_axis(0), _axis(1))) == [] + assert tracker.events == [] + + +def test_an_alarm_that_is_already_up_when_first_seen_is_reported(): + tracker = AxisEventTracker() + assert _kinds(tracker.update(_snap(_axis(2, alarm=True)))) == [(2, ev.ALARM_ON)] + + +def test_alarm_enable_and_homing_edges_are_reported_once(): + tracker = AxisEventTracker() + tracker.update(_snap(_axis(0, enabled=False))) + assert _kinds(tracker.update(_snap(_axis(0, enabled=True)))) == [(0, ev.ENABLED)] + assert _kinds(tracker.update(_snap(_axis(0, alarm=True)))) == [(0, ev.ALARM_ON)] + assert tracker.update(_snap(_axis(0, alarm=True))) == [] + assert _kinds(tracker.update(_snap(_axis(0)))) == [(0, ev.ALARM_OFF)] + assert _kinds(tracker.update(_snap(_axis(0, homing=True)))) == [(0, ev.HOME_STARTED)] + assert _kinds(tracker.update(_snap(_axis(0, home_status=1)))) == [(0, ev.HOME_DONE)] + tracker.update(_snap(_axis(0, homing=True))) + assert _kinds(tracker.update(_snap(_axis(0, home_status=0)))) == [(0, ev.HOME_INCOMPLETE)] + + +def test_a_read_failure_and_its_recovery_are_reported(): + tracker = AxisEventTracker() + tracker.update(_snap(_axis(1))) + failed = tracker.update(_snap(_axis(1, read_ok=False, error="timeout"))) + assert [(e.axis, e.kind, e.detail) for e in failed] == [(1, ev.READ_FAILED, "timeout")] + assert tracker.update(_snap(_axis(1, read_ok=False, error="timeout"))) == [] + assert _kinds(tracker.update(_snap(_axis(1)))) == [(1, ev.READ_OK)] + + +def test_a_gap_in_the_connection_resets_the_baseline(): + tracker = AxisEventTracker() + tracker.update(_snap(_axis(0))) + assert _kinds(tracker.update(_snap(connected=False))) == [(ev.NO_AXIS, ev.DISCONNECTED)] + # Back again with the axis alarmed: the connection edge, then the alarm as a + # first sight; nothing is compared with a state from before the gap. + back = tracker.update(_snap(_axis(0, alarm=True))) + assert _kinds(back) == [(ev.NO_AXIS, ev.CONNECTED), (0, ev.ALARM_ON)] + + +def test_units_disagreeing_with_the_configuration_is_an_event_both_ways(): + tracker = AxisEventTracker() + expected = {0: 10.0} + assert tracker.update(_snap(_axis(0, units=10.0)), expected) == [] + mismatch = tracker.update(_snap(_axis(0, units=5.0)), expected) + assert [(e.kind, e.detail) for e in mismatch] == [(ev.UNITS_MISMATCH, "10|5")] + assert tracker.update(_snap(_axis(0, units=5.0)), expected) == [] + assert _kinds(tracker.update(_snap(_axis(0, units=10.0)), expected)) == [(0, ev.UNITS_OK)] + # With nothing to compare against no verdict is invented. + assert AxisEventTracker().update(_snap(_axis(0, units=5.0))) == [] + + +def test_the_log_is_bounded_and_can_be_cleared(): + tracker = AxisEventTracker(limit=3) + tracker.update(_snap(_axis(0))) + for index in range(10): + tracker.update(_snap(_axis(0, alarm=bool(index % 2 == 0)))) + assert len(tracker.events) == 3 + tracker.clear() + assert tracker.events == [] diff --git a/apps/autocal/tests/test_device_setup.py b/apps/autocal/tests/test_device_setup.py index ceff032..340ba69 100644 --- a/apps/autocal/tests/test_device_setup.py +++ b/apps/autocal/tests/test_device_setup.py @@ -180,7 +180,7 @@ def test_the_axis_mapping_page_holds_the_mapping_and_refuses_duplicates(make_win window = make_window() try: dialog = window.device_setup - assert dialog.tabs.count() == 4 + assert dialog.tabs.count() == 5 page = dialog.roles_page assert page.roles() == DEFAULT_AXIS_ROLES assert page.problem() == "" diff --git a/apps/autocal/tests/test_unit_report.py b/apps/autocal/tests/test_unit_report.py new file mode 100644 index 0000000..88921ac --- /dev/null +++ b/apps/autocal/tests/test_unit_report.py @@ -0,0 +1,110 @@ +"""The unit table: what is configured, what the controller reads, and the verdict.""" + +from __future__ import annotations + +import math + +import pytest +from orisys_autocal.motion import AxisState, Snapshot +from orisys_autocal.motion.models import AxisStatusBits +from orisys_autocal.motion.service import DEFAULT_AXIS_ROLES +from orisys_autocal.motion.unit_report import ( + INVALID, + MISMATCH, + OK, + UNCALIBRATED, + UNMAPPED, + UNREAD, + build_unit_rows, + summarise, +) + +ROLES = dict(DEFAULT_AXIS_ROLES) # x0 y1 yaw2 pitch3 z4 + + +def _snapshot(units: dict[int, float], *, connected: bool = True) -> Snapshot: + axes = tuple( + AxisState(axis=axis, units=value, status=AxisStatusBits(0), read_ok=True) + for axis, value in units.items() + ) + return Snapshot(connected=connected, axes=axes) + + +def _rows(expected: float | dict[int, float], observed: dict[int, float] | None, **kw): + def expected_of(axis: int) -> float: + return expected[axis] if isinstance(expected, dict) else expected + + snapshot = None if observed is None else _snapshot(observed, **kw) + return {row.role: row for row in build_unit_rows(ROLES, expected_of, snapshot)} + + +def test_a_matching_platform_is_all_ok_with_the_bench_scales(): + rows = _rows(10.0, {axis: 10.0 for axis in range(5)}) + assert [row.status for row in rows.values()] == [OK] * 5 + # Linear: 5000 pulses per millimetre and a pulse equivalent of 10. + assert rows["x"].per_unit == pytest.approx(10.0 / 5000.0) + assert rows["x"].per_mm == pytest.approx(500.0) + assert rows["z"].kind == "linear" + # Rotary: 12500 and 29500 user units per quarter turn. + assert rows["yaw"].axis == 2 + assert rows["yaw"].per_mm == pytest.approx(12500.0 / 90.0) + assert rows["pitch"].per_mm == pytest.approx(29500.0 / 90.0) + assert rows["yaw"].per_unit == pytest.approx(90.0 / 12500.0) + assert [row.role for row in rows.values()] == ["x", "y", "z", "yaw", "pitch"] + + +def test_no_connection_or_no_readback_is_unread_not_fine(): + assert {row.status for row in _rows(10.0, None).values()} == {UNREAD} + offline = _rows(10.0, {axis: 10.0 for axis in range(5)}, connected=False) + assert {row.status for row in offline.values()} == {UNREAD} + assert all(row.observed is None for row in offline.values()) + + +def test_a_controller_that_disagrees_is_a_mismatch_for_that_axis_only(): + observed = {axis: 10.0 for axis in range(5)} + observed[1] = 5.0 + rows = _rows(10.0, observed) + assert rows["y"].status == MISMATCH + assert rows["y"].observed == 5.0 + assert [row.status for role, row in rows.items() if role != "y"] == [OK] * 4 + good, total, problems = summarise(list(rows.values())) + assert (good, total, [row.role for row in problems]) == (4, 5, ["y"]) + + +def test_rotary_axes_are_not_calibrated_away_from_the_bench_equivalent(): + # Same on both sides, so the controller agrees, but there is no angle scale. + rows = _rows({0: 10.0, 1: 10.0, 2: 5.0, 3: 10.0, 4: 10.0}, {0: 10.0, 1: 10.0, 2: 5.0, 3: 10.0, 4: 10.0}) + assert rows["yaw"].status == UNCALIBRATED + assert math.isnan(rows["yaw"].per_unit) and math.isnan(rows["yaw"].per_mm) + # A linear axis rescales freely: the scale follows the equivalent. + linear = _rows({0: 5.0, 1: 10.0, 2: 10.0, 3: 10.0, 4: 10.0}, {0: 5.0, 1: 10.0, 2: 10.0, 3: 10.0, 4: 10.0}) + assert linear["x"].status == OK + assert linear["x"].per_mm == pytest.approx(1000.0) + + +@pytest.mark.parametrize("bad", [0.0, -1.0, math.nan, math.inf]) +def test_an_unusable_configured_value_is_invalid(bad): + rows = _rows(bad, {axis: 10.0 for axis in range(5)}) + # inf is not finite either; every row is refused, none shows a made-up scale. + assert {row.status for row in rows.values()} == {INVALID} + assert all(math.isnan(row.per_unit) for row in rows.values()) + + +def test_a_role_without_a_drive_is_unmapped(): + roles = {key: value for key, value in ROLES.items() if key != "z"} + rows = {r.role: r for r in build_unit_rows(roles, lambda axis: 10.0, _snapshot({0: 10.0}))} + assert rows["z"].status == UNMAPPED + assert rows["z"].axis == -1 + + +def test_the_verdict_agrees_with_the_motion_guard(): + """The page must never say fine where :meth:`MotionService.units_problem` refuses.""" + from orisys_autocal.motion.service import MotionService + + service = MotionService() + observed = {axis: 10.0 for axis in range(5)} + snapshot = _snapshot(observed) + service._last_axes = {state.axis: state for state in snapshot.axes} # noqa: SLF001 + rows = {r.role: r for r in build_unit_rows(service.axis_roles, lambda a: service.params(a).units, snapshot)} + for role, row in rows.items(): + assert (service.units_problem(role, row.axis) == "") == row.ok, role diff --git a/apps/autocal/tests/test_units_page.py b/apps/autocal/tests/test_units_page.py new file mode 100644 index 0000000..cd7cea1 --- /dev/null +++ b/apps/autocal/tests/test_units_page.py @@ -0,0 +1,213 @@ +"""The units page (locked by default) and the axis page's alarm and event log. + +Interface tests: they drive the widgets an operator uses. The controller is never +reached; snapshots are built by hand, and ``QMessageBox.question`` is answered by the +test, so nothing here opens a dialog. +""" + +from __future__ import annotations + +import pytest +from orisys_app_shared.i18n import t +from orisys_autocal.motion import AxisState, Snapshot +from orisys_autocal.motion.models import AxisStatusBits +from orisys_autocal.motion.service import DEFAULT_AXIS_ROLES +from orisys_autocal.motion.unit_report import build_unit_rows +from PySide6.QtWidgets import QMessageBox + +ROLES = dict(DEFAULT_AXIS_ROLES) + + +def _snapshot(*, connected: bool = True, alarm_axis: int | None = None, units: float = 10.0): + axes = tuple( + AxisState( + axis=axis, + enabled=True, + units=units, + status=AxisStatusBits(512 if axis == alarm_axis else 0), + read_ok=True, + ) + for axis in range(5) + ) + return Snapshot(connected=connected, axes=axes) + + +def _rows(connected: bool = True, expected: float = 10.0): + return build_unit_rows(ROLES, lambda axis: expected, _snapshot(connected=connected)) + + +def _answer(monkeypatch, button) -> list[str]: + """Answer the confirmation with ``button`` and record what it asked.""" + asked: list[str] = [] + + def question(parent, title, body, *args, **kwargs): + asked.append(body) + return button + + monkeypatch.setattr(QMessageBox, "question", question) + return asked + + +def test_the_page_is_locked_until_the_operator_presses_modify(make_window): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(connected=False), False) + assert page.locked + assert all(edit.isReadOnly() for edit in page._edits.values()) # noqa: SLF001 + # Nothing to change on a machine that is not connected. + assert not page.modify_button.isEnabled() + + page.set_rows(_rows(), True) + assert page.modify_button.isEnabled() + page.modify_button.click() + assert not page.locked + assert not any(edit.isReadOnly() for edit in page._edits.values()) # noqa: SLF001 + assert page.apply_button.isEnabled() + window.close() + + +def test_a_confirmed_change_is_emitted_and_the_page_locks_again(make_window, monkeypatch): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + emitted: list[dict] = [] + page.unitsApplied.connect(emitted.append) + asked = _answer(monkeypatch, QMessageBox.StandardButton.Yes) + + page.modify_button.click() + page._edits["x"].setText("5") # noqa: SLF001 + page.apply_button.click() + + assert emitted == [{0: 5.0}] + assert page.locked + assert "5" in asked[0] and "10" in asked[0] + window.close() + + +def test_declining_the_confirmation_changes_nothing(make_window, monkeypatch): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + emitted: list[dict] = [] + page.unitsApplied.connect(emitted.append) + _answer(monkeypatch, QMessageBox.StandardButton.No) + + page.modify_button.click() + page._edits["y"].setText("20") # noqa: SLF001 + page.apply_button.click() + + assert emitted == [] + assert not page.locked # still the operator's to finish or cancel + page.cancel_button.click() + assert page.locked + assert page._edits["y"].text() == "10" # noqa: SLF001 + window.close() + + +def test_a_rotary_value_off_the_calibration_is_warned_about(make_window, monkeypatch): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + asked = _answer(monkeypatch, QMessageBox.StandardButton.No) + + page.modify_button.click() + page._edits["yaw"].setText("5") # noqa: SLF001 + page.apply_button.click() + + assert t("units.confirm_rotary", roles=t("setup.role_yaw"), cal="10") in asked[0] + window.close() + + +@pytest.mark.parametrize("bad", ["", "abc", "0", "-3", "nan", "inf"]) +def test_an_unusable_value_is_refused_before_any_question(make_window, monkeypatch, bad): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + emitted: list[dict] = [] + page.unitsApplied.connect(emitted.append) + asked = _answer(monkeypatch, QMessageBox.StandardButton.Yes) + + page.modify_button.click() + page._edits["z"].setText(bad) # noqa: SLF001 + page.apply_button.click() + + assert emitted == [] and asked == [] + assert t("units.invalid", role=t("setup.role_z")) == page.message_label.text() + window.close() + + +def test_applying_without_a_change_just_locks(make_window, monkeypatch): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + emitted: list[dict] = [] + page.unitsApplied.connect(emitted.append) + asked = _answer(monkeypatch, QMessageBox.StandardButton.Yes) + + page.modify_button.click() + page.apply_button.click() + + assert emitted == [] and asked == [] + assert page.locked + window.close() + + +def test_a_changed_axis_mapping_drops_an_open_edit(make_window): + window = make_window() + page = window.device_setup.units_page + page.set_rows(_rows(), True) + page.modify_button.click() + page._edits["x"].setText("5") # noqa: SLF001 + + swapped = dict(ROLES, x=1, y=0) + page.set_rows(build_unit_rows(swapped, lambda axis: 10.0, _snapshot()), True) + + assert page.locked + assert page._edits["x"].text() == "10" # noqa: SLF001 + window.close() + + +def test_the_units_tab_and_the_axis_page_are_wired(make_window, monkeypatch): + window = make_window() + monkeypatch.setattr(window.poller, "submit", lambda *args, **kwargs: 0) + panel = window.axis_panel + # The always-live form no longer takes a pulse equivalent. + assert panel.units_edit.isReadOnly() + # The button on the axis page opens the units tab. + panel.unitsRequested.emit() + assert window.device_setup.current_tab() == "units" + # A confirmed change goes out as an ordinary axis-parameter apply. + applied: list[tuple] = [] + panel.paramsApplied.connect(lambda axis, params: applied.append((axis, params.units))) + window.device_setup.units_page.unitsApplied.emit({1: 5.0}) + assert applied == [(1, 5.0)] + assert panel.expected_units(1) == 5.0 + window.close() + + +def test_the_axis_page_logs_alarms_and_can_clear_the_log(make_window): + window = make_window() + panel = window.axis_panel + panel.update_snapshot(_snapshot()) + assert panel.event_log.count() == 0 + assert not panel.events_empty.isHidden() + + panel.update_snapshot(_snapshot(alarm_axis=2)) + assert panel.event_log.count() == 1 + assert t("axes.ev_alarm_on", axis=2) in panel.event_log.item(0).text() + assert panel.events_empty.isHidden() + + panel.clear_events_button.click() + assert panel.event_log.count() == 0 + window.close() + + +def test_a_unit_mismatch_reaches_the_axis_page_summary(make_window): + window = make_window() + window.axis_panel.update_snapshot(_snapshot(units=5.0)) + window._refresh_unit_report() # noqa: SLF001 - the window does this on every poll + assert t("units.summary_bad", bad=5, total=5) == window.axis_panel.units_chip.text() + window.axis_panel.update_snapshot(_snapshot(units=10.0)) + window._refresh_unit_report() # noqa: SLF001 + assert t("units.summary_ok", total=5) == window.axis_panel.units_chip.text() + window.close()