added ruff test

This commit is contained in:
ykcheungab
2026-09-27 19:20:43 +08:00
parent bcab2790bf
commit fa2703968e
178 changed files with 3277 additions and 1034 deletions
+4 -2
View File
@@ -66,14 +66,16 @@ def main(argv: list[str] | None = None) -> int:
args = _parse_args(list(argv if argv is not None else sys.argv[1:]))
_force_utf8_streams()
from orisys_app_shared import API_VERSION, __version__ as shared_version
from orisys_app_shared import API_VERSION
from orisys_app_shared import __version__ as shared_version
from orisys_app_shared.compat import check_shared_compat
from orisys_app_shared.i18n import set_locale
from orisys_app_shared.logging import init_logging, install_exception_hooks
from orisys_app_shared.theme import build_qss
from PySide6.QtWidgets import QApplication
from . import REQUIRES_SHARED_API, __version__ as app_version
from . import REQUIRES_SHARED_API
from . import __version__ as app_version
from .motion import ConnectionSpec, DemoTransport
from .preferences import load_preferences
from .ui.window import AutoCalMainWindow
@@ -16,7 +16,22 @@ from .models import (
Snapshot,
)
from .poller import MotionPoller
from .service import DEFAULT_BAS_TASK, PLATFORM_AXES, MotionService, clamp_bas_task
from .pose_move import (
MovePlan,
MoveStep,
PoseMover,
plan_pose_move,
plan_return_to_initial,
)
from .service import (
AXIS_ROLES,
DEFAULT_AXIS_ROLES,
DEFAULT_BAS_TASK,
PLATFORM_AXES,
MotionService,
clamp_bas_task,
normalize_axis_roles,
)
from .transport import DemoTransport, NativeTransport, ZmcTransport
from .units import (
DEFAULT_ACCEL,
@@ -33,7 +48,9 @@ from .units import (
)
__all__ = [
"AXIS_ROLES",
"DEFAULT_ACCEL",
"DEFAULT_AXIS_ROLES",
"DEFAULT_BAS_TASK",
"DEFAULT_SPEED",
"DEFAULT_UNITS",
@@ -55,12 +72,18 @@ __all__ = [
"MotionPoller",
"MotionResult",
"MotionService",
"MovePlan",
"MoveStep",
"NativeTransport",
"PoseMover",
"Snapshot",
"ZmcTransport",
"clamp_bas_task",
"deg_to_user_units",
"mm_to_user_units",
"normalize_axis_roles",
"plan_pose_move",
"plan_return_to_initial",
"safe_float",
"units_per_degree",
"user_units_to_deg",
@@ -438,7 +438,7 @@ class PlatformStack:
def alignment_branches(
self, unit: np.ndarray, tool: np.ndarray
) -> list["Branch"]:
) -> list[Branch]:
"""Every exact pose that puts ``unit`` on ``tool``, in closed form.
Two ``atan2`` calls, no search. The chain is yaw-then-pitch, so:
@@ -583,7 +583,7 @@ class PlatformStack:
return math.degrees(math.atan2(sine_term, cosine_term))
@staticmethod
def select_branch(branches: list["Branch"]) -> "Branch":
def select_branch(branches: list[Branch]) -> Branch:
"""Pick from :meth:`alignment_branches`: reachable, then upright, then best.
Every branch aligns to within floating point, so alignment itself rarely
@@ -596,7 +596,7 @@ class PlatformStack:
stages is to align. The tie only decides between poses that are the same
to any tolerance the machine can hold.
"""
def rank(branch: "Branch") -> tuple:
def rank(branch: Branch) -> tuple:
# Alignment first: if one branch genuinely aligns better, that is the
# answer, because aligning is what the stages are for. Reachability and
# orientation decide between branches that align the same, which is the
@@ -1051,7 +1051,7 @@ class PlatformStack:
def _better_pose(
candidate: tuple[float, float, float],
best: tuple[float, float, float],
stack: "PlatformStack",
stack: PlatformStack,
) -> bool:
"""Rank two ``(residual, yaw, pitch)`` candidates.
@@ -173,14 +173,14 @@ class MotionResult:
message: str = ""
@classmethod
def success(cls, message: str = "") -> "MotionResult":
def success(cls, message: str = "") -> MotionResult:
return cls(ok=True, code=0, message=message)
@classmethod
def refused(cls, message: str) -> "MotionResult":
def refused(cls, message: str) -> MotionResult:
"""The command was not sent: a guard rejected it before any motion."""
return cls(ok=False, code=0, message=message)
@classmethod
def failed(cls, code: int, message: str) -> "MotionResult":
def failed(cls, code: int, message: str) -> MotionResult:
return cls(ok=False, code=int(code), message=message)
@@ -21,7 +21,8 @@ Contract:
from __future__ import annotations
import logging
from typing import Any, Callable, Iterable
from collections.abc import Callable, Iterable
from typing import Any
from PySide6.QtCore import (
QMetaObject,
@@ -288,7 +289,6 @@ class MotionPoller(QObject):
# closes immediately and leaves the platform coasting.
service.stop_all()
service.disconnect()
return None
return self.submit(_disconnect)
@@ -0,0 +1,315 @@
"""Moving the platform to a simulated pose, one axis at a time.
The finger screen solves a pose — rotate the stages, slide the table — and shows it.
Executing it is a *sequence* of coordinated moves, and where that sequence lives
matters:
* The **service** owns the guards: connected, bus initialised, mapping sane, targets
finite, inside the P0 limits, rotary axes calibrated. Nothing here bypasses them.
* The **worker thread** owns the controller calls. One command per poll tick is how
:class:`~orisys_autocal.motion.poller.MotionPoller` is built, and it is why a stop
is possible at all: a mover that sat in a loop inside one command could not be
interrupted, because the stop would have to wait for the loop it is meant to abort.
* **This module** owns the order and the waiting. It watches the snapshots the poller
already produces, issues the next axis when the current one has arrived, and gives
up on its own if a step stops making progress.
That shape is the whole reason for the class: ``stop`` is submitted as an ordinary
command, so it runs at the next tick and stops the axis that is moving *now*.
Why the order is what it is: the model composes ``R_pitch · R_yaw``, i.e. the yaw acts
on the body first. Commanding yaw before pitch keeps every intermediate pose on the
path the operator watched in the preview. Table X/Y last, because the table
translation is what the rotations are resolved against.
"""
from __future__ import annotations
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from PySide6.QtCore import QObject, Signal
from .models import MotionResult, Snapshot
from .service import LINEAR_ROLES, MotionService
from .units import user_units_to_deg
#: How close an axis must be to its target before the next one starts, per unit:
#: millimetres for the linear roles, degrees for the rotary ones. One user unit is
#: 0.2 µm on the bench (``units=10``, 5000 pulses/mm), so 0.01 mm is 50 units —
#: loose enough that a drive which reports its final position one tick late does not
#: stall the sequence, tight enough that a missed target is caught.
SETTLE_MM = 0.01
SETTLE_DEG = 0.01
#: Per-step patience. A step that is neither arriving nor moving this long is
#: treated as stuck: reported and abandoned, with no further axes commanded.
STEP_TIMEOUT_S = 20.0
#: How long a step may report itself idle while still short of its target before it
#: counts as failed — a drive that has stopped without arriving will never arrive.
STALL_GRACE_S = 2.0
#: The order a role move is issued in, for the reason in the module docstring.
MOVE_ORDER: tuple[str, ...] = ("yaw", "pitch", "x", "y", "z")
#: The order "return to the initial state" retracts in: the press axis first, so the
#: tool is clear of the sensor before anything rotates, then the stages, then the
#: table. Returning in one coordinated swoop would move five axes through poses
#: nobody has simulated.
RETURN_ORDER: tuple[str, ...] = ("z", "pitch", "yaw", "x", "y")
@dataclass(frozen=True)
class MoveStep:
"""One axis, one target, in that role's own unit."""
role: str
axis: int
target: float
units: float
unit: str
def label(self) -> str:
return f"{self.role} {self.target:+.4f} {self.unit}"
@dataclass(frozen=True)
class MovePlan:
"""A sequence of steps, and what it is for."""
steps: tuple[MoveStep, ...]
title: str
#: Roles deliberately left alone. Reported, because "z did not move" is
#: information the operator needs, not an omission.
held: tuple[str, ...] = ()
def describe(self) -> str:
parts = [step.label() for step in self.steps]
text = " · ".join(parts) if parts else "-"
if self.held:
text += f" · {', '.join(self.held)} 不动作"
return text
@dataclass
class MoveState:
"""What the mover is doing, for the panel to show."""
plan: MovePlan | None = None
index: int = 0
started_at: float = 0.0
finished: bool = False
failure: str = ""
cancelled: bool = False
_step_started_at: float = field(default=0.0)
_idle_since: float | None = None
@property
def step(self) -> MoveStep | None:
if self.plan is None or not 0 <= self.index < len(self.plan.steps):
return None
return self.plan.steps[self.index]
@property
def total(self) -> int:
return len(self.plan.steps) if self.plan else 0
def plan_pose_move(
service: MotionService,
targets: Mapping[str, float],
*,
title: str = "",
hold: Sequence[str] = (),
order: Sequence[str] = MOVE_ORDER,
) -> MovePlan:
"""Turn per-role targets into a stepped plan, in ``order``.
``targets`` is in each role's own unit. Roles absent from it are not commanded —
that is how "z does not move" is expressed: not by a target equal to its current
position, which would still command it.
"""
steps: list[MoveStep] = []
for role in order:
if role not in targets:
continue
axis = service.role_axis(role)
if axis is None:
continue
value = float(targets[role])
steps.append(
MoveStep(
role=role,
axis=axis,
target=value,
units=service.to_units(role, value),
unit="mm" if role in LINEAR_ROLES else "°",
)
)
return MovePlan(steps=tuple(steps), title=title, held=tuple(hold))
def plan_return_to_initial(service: MotionService) -> MovePlan:
"""Every axis to zero, press axis first, in :data:`RETURN_ORDER`."""
return plan_pose_move(
service,
{role: 0.0 for role in RETURN_ORDER},
order=RETURN_ORDER,
)
class PoseMover(QObject):
"""Issues a :class:`MovePlan` one step at a time, off the poller's snapshots."""
started = Signal(object)
stepStarted = Signal(object)
finished = Signal(object)
failed = Signal(str)
cancelled = Signal()
def __init__(self, service: MotionService, submit, parent: QObject | None = None) -> None:
"""
Args:
service: the motion service, for unit conversion and per-axis guards.
submit: the poller's ``submit``, taking a ``service -> result`` callable.
"""
super().__init__(parent)
self._service = service
self._submit = submit
self.state = MoveState()
# ------------------------------------------------------------------ control
@property
def running(self) -> bool:
plan = self.state.plan
return bool(
plan is not None
and not self.state.finished
and not self.state.failure
and not self.state.cancelled
)
def start(self, plan: MovePlan, *, now: float | None = None) -> MotionResult:
"""Validate and begin. Refusal is returned, not raised."""
if self.running:
return MotionResult.refused("已有运动在执行 / a move is already running")
refusal = self._service.check_axes_move(
{step.role: step.target for step in plan.steps}
)
if refusal is not None:
return refusal
if not plan.steps:
return MotionResult.refused("没有要移动的轴 / nothing to move")
stamp = time.monotonic() if now is None else now
self.state = MoveState(plan=plan, started_at=stamp, _step_started_at=stamp)
self.started.emit(plan)
self._issue_current()
return MotionResult.success(plan.describe())
def cancel(self, reason: str = "") -> None:
"""Abandon the queue. The axis already moving is stopped by the caller.
Deliberately does not stop anything itself: stopping is a controller command
with its own guard, and the window issues it as one. What this guarantees is
that no *further* axis is commanded after the stop.
"""
if self.state.plan is None or self.state.finished:
return
self.state.cancelled = True
self.state.failure = reason
self.state.plan = None
self.cancelled.emit()
def _issue_current(self) -> None:
step = self.state.step
if step is None:
self._complete()
return
self.state._step_started_at = time.monotonic()
self.state._idle_since = None
if not step.units == step.units: # NaN: no calibration for this role
self._fail(f"{step.role} 无法换算成用户单位 / cannot convert {step.role}")
return
self._submit(lambda service, step=step: service.move_abs(step.axis, step.units))
self.stepStarted.emit(step)
def _complete(self) -> None:
self.state.finished = True
plan = self.state.plan
self.state.plan = None
self.finished.emit(plan)
def _fail(self, message: str) -> None:
self.state.failure = message
self.state.plan = None
self.failed.emit(message)
# ---------------------------------------------------------------- progress
def on_snapshot(self, snapshot: Snapshot) -> None:
"""Advance the plan from a poll. Called on every refresh."""
if not self.running:
return
step = self.state.step
if step is None:
self._complete()
return
now = time.monotonic()
if now - self.state._step_started_at > STEP_TIMEOUT_S:
self._fail(
f"{step.role} 等待超时 / timed out waiting for {step.role}"
)
return
state = snapshot.axis(step.axis)
if state is None:
return
if state.error or not state.read_ok:
self._fail(f"{step.role}: {state.error or 'read failed'}")
return
if state.status.alarm:
self._fail(f"{step.role} 报警 / alarm on {step.role}")
return
# A linear role reads in millimetres, a rotary one in degrees; the service
# converts lengths, the units module converts calibrated angles.
if step.unit == "mm":
travelled = self._service.to_mm(step.axis, state.dpos)
else:
travelled = user_units_to_deg(step.axis, state.dpos)
tolerance = SETTLE_MM if step.unit == "mm" else SETTLE_DEG
if not state.idle:
self.state._idle_since = None
return
if abs(travelled - step.target) <= tolerance:
self.state.index += 1
self._issue_current()
return
# Idle, but not there. A drive that has stopped short will not resume, so
# this is a failure rather than something to keep waiting for.
if self.state._idle_since is None:
self.state._idle_since = now
elif now - self.state._idle_since > STALL_GRACE_S:
self._fail(
f"{step.role} 停在 {travelled:.4f}{step.unit},目标 "
f"{step.target:.4f}{step.unit} / stopped short of target"
)
def progress_text(self) -> str:
"""A one-line status for the panel."""
if self.state.failure:
return f"✗ {self.state.failure}"
if self.state.finished:
return "✓ 已完成 / done"
step = self.state.step
if step is None:
return ""
return (
f"{self.state.index + 1}/{self.state.total} "
f"{step.role} → {step.target:+.4f} {step.unit}"
)
@@ -5,18 +5,19 @@ without a GUI and without hardware. :mod:`~orisys_autocal.motion.poller` drives
it from a worker thread; the UI only ever calls these methods.
Guards live here rather than in the UI on purpose. A motion command is refused
for a stated reason — bus not initialised, axis not idle, non-numeric input —
and :meth:`MotionService.jog` and friends return that reason as a
``MotionResult`` instead of silently doing nothing.
for a stated reason — bus not initialised, axis not idle, non-numeric input, a
target past the P0 travel limit, an axis mapping that does not add up — and
:meth:`MotionService.jog` and friends return that reason as a ``MotionResult``
instead of silently doing nothing.
"""
from __future__ import annotations
import logging
import math
import time
import os
from typing import Callable, Iterable
import time
from collections.abc import Callable, Iterable, Mapping
from orisys_app_shared.zmotion import ZAuxError
@@ -34,13 +35,14 @@ from .models import (
MotionResult,
Snapshot,
)
from .transport import DemoTransport, ZmcTransport, NativeTransport
from .transport import NativeTransport, ZmcTransport
from .units import (
DEFAULT_ACCEL,
DEFAULT_SPEED,
DEFAULT_UNITS,
PULSES_PER_MM,
deg_to_user_units,
mm_to_user_units,
safe_float,
units_per_degree,
user_units_to_mm,
)
@@ -49,6 +51,32 @@ logger = logging.getLogger(__name__)
#: The platform has five axes; the bus may report fewer if a drive is missing.
PLATFORM_AXES = 5
#: The platform directions a drive can be, in the order the model's frames are
#: named. A drive's *number* says nothing about what it moves: the bus reports five
#: identical CiA-402 axes, and whether drive 2 tilts the sensor or lifts the table is
#: a property of the bench's wiring. This is the mapping the whole "move the platform
#: to the simulated pose" path rests on, so it is configuration with a page in the
#: device setup rather than a constant.
AXIS_ROLES: tuple[str, ...] = ("x", "y", "z", "yaw", "pitch")
#: Roles whose target is a length, and roles whose target is an angle. A degree
#: target checked against a millimetre limit is not a limit at all.
LINEAR_ROLES: tuple[str, ...] = ("x", "y", "z")
ROTARY_ROLES: tuple[str, ...] = ("yaw", "pitch")
#: Default mapping, from the bench: drives 0-4 are x, y, pitch, yaw, z.
#:
#: Deliberately *not* the same as the model's own order. It is written the way the
#: machine is wired, so reading it against the terminal strip is a check rather than
#: a translation.
DEFAULT_AXIS_ROLES: dict[str, int] = {
"x": 0,
"y": 1,
"pitch": 2,
"yaw": 3,
"z": 4,
}
#: ``BUS_TYPE`` values written by the vendor ``.bas`` programs.
_BUS_TYPE_NAMES: dict[int, BusType] = {0: "ecat", 1: "rtex"}
@@ -78,6 +106,27 @@ def clamp_bas_task(value: object, fallback: int = DEFAULT_BAS_TASK) -> int:
return task if low <= task <= high else fallback
def normalize_axis_roles(roles: Mapping[str, int] | None) -> dict[str, int]:
"""A clean role -> drive mapping, falling back to the bench default.
Unusable entries are dropped rather than coerced: a role mapped to a drive the
file did not contain is a configuration the app should report through
:meth:`MotionService.roles_problem`, not silently repair.
"""
if roles is None:
return dict(DEFAULT_AXIS_ROLES)
clean: dict[str, int] = {}
for role in AXIS_ROLES:
value = roles.get(role)
if value is None:
continue
try:
clean[role] = int(value)
except (TypeError, ValueError):
continue
return clean
def _online_failure(reply: str) -> str:
"""The controller's own refusal text from an online command, or ``""``.
@@ -107,6 +156,8 @@ class MotionService:
pulses_per_mm: float = PULSES_PER_MM,
language: str = "zh_CN",
bas_task: int = DEFAULT_BAS_TASK,
axis_roles: Mapping[str, int] | None = None,
limits: Mapping[str, float] | None = None,
) -> None:
self._transport: ZmcTransport = transport or NativeTransport()
self.axis_count = int(axis_count)
@@ -114,6 +165,14 @@ class MotionService:
self._pulses_per_mm = pulses_per_mm
self.language = language
self.bas_task = clamp_bas_task(bas_task)
# Which drive is which platform direction. The app cannot infer this from a
# CiA-402 axis: the bus reports numbers. It is configuration, so it arrives
# from the preference file and can be corrected in the device setup.
self._axis_roles: dict[str, int] = normalize_axis_roles(axis_roles)
# The mm limits from workflow P0. They live here rather than in the UI because
# this is where motion is allowed or refused, and a limit the UI enforces is a
# limit a later caller can walk around.
self._limits: dict[str, float] = dict(limits or {})
self._spec: ConnectionSpec | None = None
self._bus = BusStatus()
@@ -151,6 +210,152 @@ class MotionService:
def params(self, axis: int) -> AxisParams:
return self._params.get(axis, AxisParams())
# --------------------------------------------------------------- axis roles
@property
def axis_roles(self) -> dict[str, int]:
"""Drive index per platform direction, as configured."""
return dict(self._axis_roles)
def set_axis_roles(self, roles: Mapping[str, int] | None) -> None:
"""Replace the mapping. Called when the device setup is edited."""
self._axis_roles = normalize_axis_roles(roles)
def set_limits(self, limits: Mapping[str, float] | None) -> None:
"""Replace the P0 limits. Called when the preference file is reloaded."""
self._limits = dict(limits or {})
def role_axis(self, role: str) -> int | None:
"""The drive that moves ``role``, or ``None`` if it is not configured."""
return self._axis_roles.get(role)
def roles_problem(self) -> str:
"""Why the mapping cannot be used, or ``""`` when it is sound.
A duplicated drive is the dangerous case — two roles claiming axis 3 means
one of them silently does not move, and the pose would be wrong rather than
refused — so it is reported rather than tolerated.
"""
missing = [role for role in AXIS_ROLES if role not in self._axis_roles]
if missing:
return self._text(
f"轴映射未配置:{', '.join(missing)}",
f"Axis mapping is not configured for: {', '.join(missing)}",
)
used: dict[int, str] = {}
for role in AXIS_ROLES:
axis = self._axis_roles[role]
if not self._known_axis(axis):
return self._text(
f"{role} 指向轴 {axis},超出 0-{self.axis_count - 1}",
f"{role} points at axis {axis}, outside 0-{self.axis_count - 1}",
)
if axis in used:
return self._text(
f"轴 {axis} 同时是 {used[axis]} 和 {role}",
f"Axis {axis} is claimed by both {used[axis]} and {role}",
)
used[axis] = role
return ""
@property
def travel_limit_mm(self) -> float:
"""The P0 travel limit. ``0`` means "not set", never "unlimited"."""
return safe_float(self._limits.get("max_travel_mm"), 0.0)
def limit_for(self, role: str) -> float:
"""The P0 limit for a role, in that role's own unit.
Linear roles are limited in millimetres and rotary roles in degrees: one
number cannot bound both, and treating a degree target as millimetres is how
a 180° move passes a 100 mm check.
"""
key = "max_travel_mm" if role in LINEAR_ROLES else f"max_{role}_deg"
return safe_float(self._limits.get(key), 0.0)
def units_problem(self, role: str, axis: int) -> str:
"""Why a role cannot be commanded in its own unit, or ``""``."""
if role in LINEAR_ROLES:
return ""
scale = units_per_degree(axis)
if not math.isfinite(scale) or scale <= 0:
return self._text(
f"{role} 用轴 {axis},但该轴没有角度标定(用户单位/度)",
f"{role} uses axis {axis}, which has no user-units-per-degree "
f"calibration",
)
return ""
def check_axes_move(self, targets: Mapping[str, float]) -> MotionResult | None:
"""Refuse a coordinated pose move, or return ``None`` to allow it.
``targets`` is per role, in the role's own unit: millimetres for x / y / z,
degrees for yaw / pitch.
Every reason a move can be wrong is checked *before* the first axis is
commanded. A pose move is several coordinated moves, and discovering at axis
three that axis four is out of range would leave the machine stopped
somewhere that is neither the start nor the target.
"""
if not self.connected:
return MotionResult.refused(self._text("未连接控制器", "Not connected"))
if not self._bus.initialized:
return MotionResult.refused(
self._text(
"总线未初始化完成,禁止运动",
"Bus is not initialised; motion is refused",
)
)
problem = self.roles_problem()
if problem:
return MotionResult.refused(problem)
if not targets:
return MotionResult.refused(self._text("没有要移动的轴", "Nothing to move"))
for role, value in targets.items():
if role not in AXIS_ROLES:
return MotionResult.refused(
self._text(f"未知方向 {role}", f"Unknown direction {role}")
)
if not isinstance(value, (int, float)) or not math.isfinite(float(value)):
return MotionResult.refused(
self._text(f"{role} 目标不是有效数字", f"{role} target is not a number")
)
axis = self._axis_roles[role]
units_problem = self.units_problem(role, axis)
if units_problem:
return MotionResult.refused(units_problem)
for role, value in targets.items():
limit = self.limit_for(role)
unit = "mm" if role in LINEAR_ROLES else "°"
if limit <= 0:
return MotionResult.refused(
self._text(
f"{role} 的上限未设置(P0),禁止运动",
f"The {role} limit is not set (P0); motion is refused",
)
)
if abs(float(value)) > limit:
return MotionResult.refused(
self._text(
f"{role} 目标 {float(value):.3f}{unit} 超出上限 {limit:.3f}{unit}",
f"{role} target {float(value):.3f}{unit} is past the "
f"{limit:.3f}{unit} limit",
)
)
return None
def to_units(self, role: str, value: float) -> float:
"""Controller user units for a role target, in that role's own unit."""
axis = self._axis_roles.get(role)
if axis is None:
return math.nan
if role in LINEAR_ROLES:
units = self.params(axis).units or self._units_per_mm
return mm_to_user_units(
value, units=units, pulses_per_mm=self._pulses_per_mm
)
return deg_to_user_units(axis, value)
def set_language(self, language: str) -> None:
self.language = language
+83 -5
View File
@@ -13,7 +13,12 @@ from typing import Any
from orisys_app_shared import prefs as _prefs
from orisys_app_shared.paths import preference_path
from .motion.service import DEFAULT_BAS_TASK, PLATFORM_AXES, clamp_bas_task
from .motion.service import (
DEFAULT_AXIS_ROLES,
DEFAULT_BAS_TASK,
PLATFORM_AXES,
clamp_bas_task,
)
from .motion.units import DEFAULT_ACCEL, DEFAULT_SPEED, DEFAULT_UNITS
#: Connection modes, in the order the combo offers them.
@@ -81,6 +86,10 @@ def _default_safety_limits() -> dict[str, float]:
``0`` means "not set" for the ones where a real value is required, and the
app reports that at the P0 gate rather than treating it as unlimited.
Lengths are millimetres and angles degrees, because the roles they bound are:
one number cannot bound both, and a degree target checked against a millimetre
limit is not a check.
"""
return {
"max_travel_mm": 100.0,
@@ -90,6 +99,11 @@ def _default_safety_limits() -> dict[str, float]:
#: How far the axis may descend looking for contact before the run is
#: aborted (workflow P7, no-contact abort).
"contact_search_mm": 3.0,
#: Stage envelope, from the confirmed bench values: the yaw stage turns a
#: half-turn either way, the pitch stage a full turn. These are the limits a
#: pose move is checked against before any axis is commanded.
"max_yaw_deg": 180.0,
"max_pitch_deg": 360.0,
}
@@ -104,9 +118,10 @@ def _default_sensor_setup() -> dict[str, Any]:
"sensor_type": "box",
"config_ref": "",
"geometry_id": "",
#: Which axis presses, and the probe tip offset from the platform's
#: reference point. Empty offset means "not set", which blocks P5.
"press_axis": 2,
#: The probe tip's offset from the platform's reference point. Empty offset
#: means "not set", which blocks P5. Which axis presses is **not** here: it
#: is ``axis_roles["z"]``, because one fact written twice is one fact that
#: eventually disagrees.
"probe_tip_offset_mm": [0.0, 0.0, 0.0],
"probe_tip_set": False,
#: 4x4 row-major sensor -> platform transform, plus how well it fitted.
@@ -173,6 +188,19 @@ def default_preferences() -> dict[str, Any]:
# --- workflow blocks (see apps/autocal/docs/workflow.md) --------------
#: P0: limits enforced from P2 onward.
"safety_limits": _default_safety_limits(),
#: Which drive is which platform direction, from the bench wiring: drives
#: 0-4 are x, y, pitch, yaw, z. There is no way to read this off the bus —
#: five CiA-402 axes report numbers, not names — so it is configuration, and
#: the pose-move path refuses to run until it is complete and free of
#: duplicates. It is also the *only* place "which axis presses" is written:
#: the sensor setup used to carry a second copy, which is how the two would
#: have drifted apart.
"axis_roles": dict(DEFAULT_AXIS_ROLES),
#: Ask before a pose move. On by default — the first moves on a machine are
#: the ones worth reading — and turnable off once the numbers are trusted,
#: because a prompt that cannot be dismissed unread is one that gets clicked
#: through without being read.
"move_confirm": True,
#: P3/P4: sensor under test, probe tip, sensor<->platform transform.
"sensor_setup": _default_sensor_setup(),
#: P5: grid generation and run parameters.
@@ -206,9 +234,26 @@ def default_preferences() -> dict[str, Any]:
}
#: Preference blocks whose *keys* are added to over time. Stored preferences are
#: merged one level deep, so a block saved by an older build replaces the whole
#: default block and any key added since reads as absent. For a P0 limit that is not
#: a cosmetic problem: the move is refused with "the limit is not set" while the file
#: in front of the operator says nothing about it. Filling the block key by key keeps
#: what was chosen and supplies only what was never there.
_MERGED_BLOCKS = ("connection", "safety_limits", "sensor_setup", "sequence_profile")
def load_preferences() -> dict[str, Any]:
"""Stored preferences with AutoCal's defaults filling the gaps."""
prefs = _prefs.load_preferences(default_preferences())
defaults = default_preferences()
prefs = _prefs.load_preferences(defaults)
for name in _MERGED_BLOCKS:
stored = prefs.get(name)
if not isinstance(stored, dict):
continue
block = dict(defaults.get(name) or {})
block.update(stored)
prefs[name] = block
# A stored document from an older build may miss a newly added axis block.
axis_params = prefs.setdefault("axis_params", {})
for axis, values in _default_axis_params().items():
@@ -252,6 +297,39 @@ def load_preferences() -> dict[str, Any]:
prefs["finger_tool_xy_mm"] = tool
# A hand-edited task number must not reach the controller as typed.
prefs["bas_task"] = clamp_bas_task(prefs.get("bas_task"))
prefs = _migrate_axis_roles(prefs)
return prefs
def _migrate_axis_roles(prefs: dict[str, Any]) -> dict[str, Any]:
"""Complete the axis mapping, and take the old duplicate press axis out.
``sensor_setup.press_axis`` named the pressing drive on its own. That is
``axis_roles["z"]`` now — the same fact — and a stored value is moved there when
the mapping has not been set, so a bench that had it configured keeps working.
The old key is removed either way: leaving it would keep a second answer to
"which axis presses" in the file.
"""
roles = prefs.get("axis_roles")
roles = dict(roles) if isinstance(roles, dict) else {}
sensor = prefs.get("sensor_setup")
if isinstance(sensor, dict):
legacy = sensor.pop("press_axis", None)
if legacy is not None and "z" not in roles:
try:
roles["z"] = int(legacy)
except (TypeError, ValueError):
pass
merged = dict(DEFAULT_AXIS_ROLES)
# A stored mapping wins where it is usable; the default fills what it does not
# cover, so an old document that only named the press axis still comes up with a
# complete mapping rather than a refusal.
for role, value in roles.items():
try:
merged[role] = int(value)
except (TypeError, ValueError):
continue
prefs["axis_roles"] = merged
return prefs
@@ -1,7 +1,8 @@
"""Sensor assets: export bundles, the flat UV view, and the mount frames."""
from .export_bundle import BundleError, SensorBundle, load_bundle
from .frames import AXIS_MAP, describe as describe_frames, mount_rotation, to_platform, to_sensor
from .frames import AXIS_MAP, mount_rotation, to_platform, to_sensor
from .frames import describe as describe_frames
from .uv_flat import (
DEFAULT_FLAT_SIZE,
FlatSurface,
@@ -12,8 +13,8 @@ from .uv_flat import (
__all__ = [
"AXIS_MAP",
"BundleError",
"DEFAULT_FLAT_SIZE",
"BundleError",
"FlatSurface",
"SensorBundle",
"describe_frames",
@@ -26,10 +26,11 @@ from __future__ import annotations
import json
import math
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Mapping
from typing import Any
#: The version this build writes, and the highest it will read.
FORMAT_VERSION = "1.0"
@@ -14,6 +14,9 @@ five-axis platform, acting on the wrong axis is the expensive mistake.
from __future__ import annotations
import math
from orisys_app_shared.i18n import t
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QBrush, QColor
from PySide6.QtWidgets import (
@@ -29,8 +32,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from ..motion import (
DEFAULT_ACCEL,
DEFAULT_SPEED,
@@ -582,7 +583,7 @@ def _millimetres(units: float, user_units: float) -> str:
if not units:
return "—"
millimetres = user_units_to_mm(user_units, units=units)
if millimetres != millimetres: # NaN
if math.isnan(millimetres):
return "—"
return f"{millimetres:.3f}"
@@ -1,17 +1,23 @@
"""The permanent axis readout rail.
Read-only by design. Every per-axis *command* lives in the Axis-debug tab, so
there is exactly one place a move can be issued from — two places to jog from is
how an operator moves the wrong axis. What this rail adds is the thing the
operator needs while working in any other tab: whether each drive is enabled,
where it is, and whether it is moving.
Read-only apart from four platform-wide actions. Every per-axis *command* lives in
the Axis-debug tab, so there is exactly one place a single-axis move can be issued
from — two places to jog from is how an operator moves the wrong axis. What this rail
adds is the thing the operator needs while working in any other tab: whether each
drive is enabled, where it is, and whether it is moving.
The three global actions are here rather than in the tab because they are the
ones you reach for in a hurry, and they must not require finding a tab first.
The four global actions are here rather than in a tab because they are the ones you
reach for in a hurry, or reach for *after* finishing in some other tab: enable,
stop, clear alarms, and return every axis to the initial state. They must not
require finding the right tab first.
"""
from __future__ import annotations
import math
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import load_colors
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QGridLayout,
@@ -23,9 +29,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import load_colors
from ..motion import Snapshot, user_units_to_mm
#: Columns of the compact readout: axis, enable state, position.
@@ -41,6 +44,10 @@ class AxisReadoutPanel(QGroupBox):
enableAllRequested = Signal(bool)
stopAllRequested = Signal()
clearAllRequested = Signal()
#: Return every drive to the initial state. Here, not in the finger screen,
#: because it is a machine-level action: it is wanted after a move made in *any*
#: tab, and the rail is on screen in all of them.
returnRequested = Signal()
def __init__(self, axis_count: int, theme: str = "dark", parent: QWidget | None = None) -> None:
super().__init__(parent)
@@ -72,7 +79,14 @@ class AxisReadoutPanel(QGroupBox):
self.stop_all_button.clicked.connect(self.stopAllRequested)
self.clear_all_button = QPushButton(self)
self.clear_all_button.clicked.connect(self.clearAllRequested)
for button in (self.enable_all_button, self.stop_all_button, self.clear_all_button):
self.return_button = QPushButton(self)
self.return_button.clicked.connect(self.returnRequested)
for button in (
self.enable_all_button,
self.stop_all_button,
self.return_button,
self.clear_all_button,
):
button.setProperty("rail", True)
outer.addWidget(button)
@@ -110,7 +124,12 @@ class AxisReadoutPanel(QGroupBox):
def set_connected(self, connected: bool) -> None:
self._connected = bool(connected)
for button in (self.enable_all_button, self.stop_all_button, self.clear_all_button):
for button in (
self.enable_all_button,
self.stop_all_button,
self.clear_all_button,
self.return_button,
):
button.setEnabled(self._connected)
def update_snapshot(self, snapshot: Snapshot) -> None:
@@ -138,7 +157,7 @@ class AxisReadoutPanel(QGroupBox):
millimetres = user_units_to_mm(state.dpos, units=state.units) if state.units else None
units_text = f"{state.dpos:.3f}"
if millimetres is not None and millimetres == millimetres:
if millimetres is not None and not math.isnan(millimetres):
units_text = t("axes.position_pair", value=units_text, mm=f"{millimetres:.3f}")
widgets["position"].setText(units_text)
widgets["position"].setToolTip(
@@ -166,5 +185,7 @@ class AxisReadoutPanel(QGroupBox):
self.setTitle(t("readout.title"))
self.stop_all_button.setText(t("axes.stop_all"))
self.clear_all_button.setText(t("axes.clear_alarm_all"))
self.return_button.setText(t("move.return_initial"))
self.return_button.setToolTip(t("move.return_hint"))
# The toggle's label depends on current state, so re-derive it.
self.update_snapshot(self._snapshot)
@@ -7,13 +7,12 @@ controller is reachable.
from __future__ import annotations
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QPixmap
from PySide6.QtWidgets import QFrame, QHBoxLayout, QLabel, QWidget
from orisys_app_shared.i18n import t
from orisys_app_shared.paths import logo_topbar_path
from orisys_app_shared.theme import load_colors
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QPixmap
from PySide6.QtWidgets import QFrame, QHBoxLayout, QLabel, QWidget
#: Chip tone -> colour token name.
_TONE_TOKENS = {
@@ -9,12 +9,12 @@ vendor reference guards this too, by checking ``BUS_TYPE`` before every call).
from __future__ import annotations
from orisys_app_shared.i18n import t
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QComboBox,
QGridLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
@@ -22,8 +22,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
#: Height of this panel's fields and buttons. Pinned for the same reason as the
#: axis panel's: an unconstrained QLineEdit is ~34 px, and four of them stacked
#: push this panel tall enough to steal an axis row from the panel above it.
@@ -11,6 +11,8 @@ from __future__ import annotations
from pathlib import Path
from orisys_app_shared.i18n import t
from orisys_app_shared.paths import app_root
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QGridLayout,
@@ -23,9 +25,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from orisys_app_shared.paths import app_root
from ..motion import INIT_FAILED, INIT_OK, INIT_RUNNING, BusStatus
_STATE_KEYS = {
@@ -2,6 +2,7 @@
from __future__ import annotations
from orisys_app_shared.i18n import t
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QComboBox,
@@ -15,8 +16,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from ..motion import ConnectionSpec
from ..preferences import CONNECTION_MODES, DEFAULT_CONTROLLER_IPS
@@ -1,9 +1,11 @@
"""The device setup: connection and bus initialisation, behind one button.
"""The device setup: connection, bus initialisation and the axis mapping.
Both groups are touched once per machine start-up — reach the controller, download
the bus program, watch the bus come up — and then not again for the rest of the
session. In the left rail they were spending ~460 px of its height on that, above
the axis readout that is read all day, which is what made the rail scroll.
These are the facts about the machine itself, as opposed to the job: reach the
controller, download the bus program, watch the bus come up, and say which drive moves
which platform direction. All of them are touched once per machine start-up — or once
per re-wire — and then not again for the rest of the session. In the left rail the
first two were spending ~460 px of its height, above the axis readout that is read all
day, which is what made the rail scroll.
They live in a **modeless dialog** rather than a modal one on purpose: connecting and
downloading are asynchronous (the poller runs the state machine, the status strip
@@ -12,35 +14,156 @@ the axis readout while it happens. Modeless also means the dialog can be opened
several places — the strip's button, its state chips, the File menu — without any of
them having to block.
The dialog owns the two panels; the window keeps a reference to each
The dialog owns the connection and bus panels; the window keeps a reference to each
(``window.connection_panel`` / ``window.bus_panel``) because it is the window that
wires them to the controller, and because the tools and checks reach for them there.
The axis mapping is read back by the window on every change, because the motion guards
have to see the same numbers the page shows.
"""
from __future__ import annotations
from orisys_app_shared.i18n import t
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QComboBox,
QDialog,
QDialogButtonBox,
QFormLayout,
QGroupBox,
QLabel,
QTabWidget,
QVBoxLayout,
QWidget,
)
from orisys_app_shared.i18n import t
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
#: Tab identity -> locale key, in the order they are shown. The order is the work:
#: reach the machine, then bring its bus up.
#: reach the machine, bring its bus up, then say what each drive moves.
TABS: tuple[tuple[str, str], ...] = (
("connection", "setup.tab_connection"),
("bus", "setup.tab_bus"),
("axes", "setup.tab_axes"),
)
#: Role -> locale key for the mapping rows.
_ROLE_KEYS = {
"x": "setup.role_x",
"y": "setup.role_y",
"z": "setup.role_z",
"yaw": "setup.role_yaw",
"pitch": "setup.role_pitch",
}
class AxisRolesPage(QWidget):
"""Which drive moves which platform direction.
A page of its own because it is a separate fact: connection and bus are about
*reaching* the machine, this is about what its drives do. Getting it wrong moves
the wrong motor, so a duplicate is reported rather than quietly accepted — and the
motion guards read the same mapping, so a page left in a contradictory state
refuses motion instead of guessing which claim was meant.
"""
rolesChanged = Signal()
def __init__(
self, roles: dict[str, int] | None = None, parent: QWidget | None = None
) -> None:
super().__init__(parent)
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)
self.box = QGroupBox(self)
self.box.setObjectName("groupBox")
form = QFormLayout(self.box)
form.setContentsMargins(10, 8, 10, 10)
form.setSpacing(6)
self.combos: dict[str, QComboBox] = {}
self.labels: dict[str, QLabel] = {}
for role in AXIS_ROLES:
label = QLabel(self.box)
combo = QComboBox(self.box)
for axis in range(PLATFORM_AXES):
combo.addItem(str(axis), axis)
combo.currentIndexChanged.connect(self._on_changed)
self.labels[role] = label
self.combos[role] = combo
form.addRow(label, combo)
column.addWidget(self.box)
self.problem_label = QLabel(self)
self.problem_label.setObjectName("statusValue")
self.problem_label.setWordWrap(True)
column.addWidget(self.problem_label)
column.addStretch(1)
self.set_roles(roles or {})
self.retranslate()
# ------------------------------------------------------------------- state
def set_roles(self, roles: dict[str, int]) -> None:
"""Show a mapping without announcing a change the operator did not make.
An empty mapping falls back to the bench default rather than to "every role
on drive 0", which would open the page in a contradictory state and — because
the guards read the same mapping — refuse motion for a reason nobody typed.
"""
source = dict(roles) or dict(DEFAULT_AXIS_ROLES)
for role, combo in self.combos.items():
value = source.get(role)
index = combo.findData(int(value)) if value is not None else -1
combo.blockSignals(True)
try:
combo.setCurrentIndex(index if index >= 0 else 0)
finally:
combo.blockSignals(False)
self._refresh_problem(announce=False)
def roles(self) -> dict[str, int]:
return {role: int(combo.currentData()) for role, combo in self.combos.items()}
def problem(self) -> str:
"""The duplicate mapping, described, or ``""`` when it is sound."""
seen: dict[int, str] = {}
for role, combo in self.combos.items():
axis = int(combo.currentData())
if axis in seen:
return t("setup.axes_duplicate", detail=f"{seen[axis]} ↔ {role}")
seen[axis] = role
return ""
def _on_changed(self, _index: int) -> None:
self._refresh_problem()
def _refresh_problem(self, *, announce: bool = True) -> None:
problem = self.problem()
self.problem_label.setText(f"⚠ {problem}" if problem else "")
if announce:
self.rolesChanged.emit()
# ------------------------------------------------------------- translations
def retranslate(self) -> None:
self.hint_label.setText(t("setup.axes_hint"))
self.box.setTitle(t("setup.tab_axes"))
for role, label in self.labels.items():
label.setText(t(_ROLE_KEYS[role]))
self._refresh_problem(announce=False)
class DeviceSetupDialog(QDialog):
"""Connection and bus initialisation, the once-per-start controls."""
@@ -49,6 +172,7 @@ class DeviceSetupDialog(QDialog):
self,
spec: ConnectionSpec,
bas_path: str = "",
roles: dict[str, int] | None = None,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
@@ -70,6 +194,7 @@ class DeviceSetupDialog(QDialog):
self.tabs.setObjectName("setupTabs")
self.connection_panel = ConnectionPanel(spec, self.tabs)
self.bus_panel = BusPanel(bas_path, self.tabs)
self.roles_page = AxisRolesPage(roles, self.tabs)
for key, locale_key in TABS:
self.tabs.addTab(self.panel(key), "")
column.addWidget(self.tabs, 1)
@@ -90,8 +215,10 @@ class DeviceSetupDialog(QDialog):
# ----------------------------------------------------------------- content
def panel(self, key: str) -> QWidget:
"""One of the two panels, by tab key."""
return self.connection_panel if key == "connection" else self.bus_panel
"""One of the three pages, by tab key."""
if key == "connection":
return self.connection_panel
return self.bus_panel if key == "bus" else self.roles_page
def open_for(self, tab: str | None = None) -> None:
"""Show the dialog, raising it if it is already up, and pick a tab.
@@ -155,3 +282,4 @@ class DeviceSetupDialog(QDialog):
self.close_button.setText(t("setup.close"))
self.connection_panel.retranslate()
self.bus_panel.retranslate()
self.roles_page.retranslate()
@@ -34,7 +34,7 @@ from orisys_app_shared.i18n import t
from ..motion.kinematics import PlatformStack, Pose, rotation_about
from ..sensor.export_bundle import SensorBundle
from ..sensor.frames import mount_rotation, to_platform, to_sensor
from ..sensor.frames import to_sensor
#: Orbit drag sensitivity and zoom step, matching the design repo's preview.
ORBIT_RAD_PER_PX = 0.008
@@ -22,6 +22,7 @@ import math
from pathlib import Path
import numpy as np
from orisys_app_shared.i18n import t
from PySide6.QtCore import QPoint, Qt, Signal
from PySide6.QtGui import QColor, QImage, QPainter, QPen, QPixmap
from PySide6.QtWidgets import (
@@ -34,7 +35,6 @@ from PySide6.QtWidgets import (
QLabel,
QLineEdit,
QPushButton,
QScrollArea,
QSizePolicy,
QTableWidget,
QTableWidgetItem,
@@ -42,8 +42,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from ..motion.kinematics import PlatformStack
from ..sensor.export_bundle import SensorBundle
from ..sensor.frames import describe as describe_frames
@@ -226,6 +224,12 @@ class FingerPanel(QGroupBox):
loadRequested = Signal()
pickRequested = Signal(int, int)
stackChanged = Signal()
#: Run the selected point's solved pose on the platform, and stop moving.
executeRequested = Signal()
stopRequested = Signal()
#: Whether to ask before moving. Persisted; the window honours it for every
#: pose move, including the rail's return-to-zero.
confirmChanged = Signal(bool)
#: Columns of the point table, minus the leading select checkbox. The first six
#: are the ones an operator may edit — the sensor-frame point and its normal —
@@ -532,8 +536,35 @@ class FingerPanel(QGroupBox):
# colour call hid the base plane and the whole trajectory without a word.
self.preview.problemChanged.connect(lambda _message: self._refresh_reach())
controls.addWidget(self.sim_button)
# Running the selected pose on the machine, and stopping it. Both live here
# because both are about *this* point's move; the stop is a second access
# point to the same command the status strip's 全部停止 issues, not a second
# implementation — an emergency stop has to be where the eyes already are.
self.execute_button = QPushButton(box)
self.execute_button.setEnabled(False)
self.execute_button.clicked.connect(self.executeRequested)
self.stop_button = QPushButton(box)
self.stop_button.setObjectName("dangerButton")
self.stop_button.setEnabled(False)
self.stop_button.clicked.connect(self.stopRequested)
controls.addWidget(self.execute_button)
controls.addWidget(self.stop_button)
# Confirming every move is right when the machine is new to the operator and
# tedious once it is not — and a prompt that cannot be turned off is one that
# gets clicked through without reading. The choice is the operator's, and the
# same policy covers the rail's return-to-zero.
self.confirm_check = QCheckBox(box)
self.confirm_check.setChecked(True)
self.confirm_check.toggled.connect(self.confirmChanged)
controls.addWidget(self.confirm_check)
controls.addStretch(1)
column.addLayout(controls)
# What the move is doing: which axis, which step, or why it refused. A move
# that reports nothing is indistinguishable from one that never started.
self.move_label = QLabel(box)
self.move_label.setObjectName("hintLabel")
self.move_label.setWordWrap(True)
column.addWidget(self.move_label)
# A real label, not text painted onto the GL widget: over a GL surface Qt
# composites a long overflowing string as an overlap of itself, and the
# legend was unreadable on the bench because of it. A label wraps.
@@ -688,6 +719,57 @@ class FingerPanel(QGroupBox):
t("finger.sim_stop") if running else t("finger.sim_play")
)
# --------------------------------------------------------------- pose move
def set_move_available(self, available: bool, *, moving: bool = False) -> None:
"""Mirror the mover's state onto the two buttons.
The stop is deliberately *not* gated on the same conditions as the execute
button: it is enabled whenever the platform may be commanded *or* something
is moving. A stop greyed out while an axis coasts would be the worst possible
button, and the emergency case is exactly the one where the execute
conditions no longer hold.
"""
self.execute_button.setEnabled(bool(available) and not moving)
self.stop_button.setEnabled(bool(available) or bool(moving))
def set_move_status(self, text: str) -> None:
"""One line about the move: step, progress, or the refusal."""
self.move_label.setText(text)
def set_confirm(self, confirm: bool) -> None:
"""Show the stored policy without announcing it as a change."""
self.confirm_check.blockSignals(True)
try:
self.confirm_check.setChecked(bool(confirm))
finally:
self.confirm_check.blockSignals(False)
def solved_pose(self) -> tuple[dict[str, float], tuple[str, ...]] | None:
"""The selected point's per-role targets, and the roles held back.
The two numbers that matter are the *table's* move and the stage angles: the
table translation is what the stages are resolved against, and it is what the
preview drew as the cyan path's end.
z is returned as *held*: executing a simulated pose moves the stages and the
table, it does not press. Pressing is P7 and has its own guards — contact
detection, press depth, the sensor's second opinion — none of which exist
here, so a move that reached the contact height would be a press with none of
them.
"""
pose = self._last_pose
if pose is None:
return None
table_x, table_y = pose.table_xy_mm
targets = {
"x": float(table_x),
"y": float(table_y),
"yaw": float(pose.yaw_deg),
"pitch": float(pose.pitch_deg),
}
return targets, ("z",)
def _sync_sim_button(self) -> None:
self.sim_button.setText(
t("finger.sim_stop") if self.preview.simulating else t("finger.sim_play")
@@ -1181,6 +1263,12 @@ class FingerPanel(QGroupBox):
self.trajectory_check.setText(t("finger.show_trajectory"))
self.sim_button.setToolTip(t("finger.sim_hint"))
self._sync_sim_button()
self.execute_button.setText(t("move.execute"))
self.execute_button.setToolTip(t("move.execute_hint"))
self.stop_button.setText(t("move.stop"))
self.stop_button.setToolTip(t("move.stop_hint"))
self.confirm_check.setText(t("move.confirm_check"))
self.confirm_check.setToolTip(t("move.confirm_hint"))
self.table.setHorizontalHeaderLabels(
[""] + [t(f"finger.col_{key}") for key in self._headers]
)
@@ -25,9 +25,11 @@ list so a runner cannot start from a table that was never validated.
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import Any
from orisys_app_shared.i18n import t
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QComboBox,
@@ -44,8 +46,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from ..preferences import POINT_ORDERS, SURFACE_MODES
from ..sequence_io import (
SequencePoint,
@@ -642,7 +642,7 @@ def _as_number(text: str, fallback: float) -> float:
value = float(str(text).strip())
except (TypeError, ValueError):
return fallback
return value if value == value else fallback # reject NaN
return value if not math.isnan(value) else fallback
def _format_number(value: Any) -> str:
@@ -20,19 +20,18 @@ from __future__ import annotations
import time
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import load_colors
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QFrame,
QHBoxLayout,
QLabel,
QPushButton,
QProgressBar,
QPushButton,
QWidget,
)
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import load_colors
from ..motion import Snapshot
#: Tone -> colour token. Matches the brand bar's chip vocabulary.
+183 -5
View File
@@ -15,12 +15,11 @@ import sys
from pathlib import Path
import numpy as np
from orisys_app_shared import API_VERSION, __version__ as shared_version
from orisys_app_shared import API_VERSION
from orisys_app_shared import __version__ as shared_version
from orisys_app_shared.i18n import set_locale, t
from orisys_app_shared.paths import outputs_dir
from orisys_app_shared.theme import build_qss, load_colors
from PySide6.QtCore import Qt, QTimer
from PySide6.QtGui import QAction, QCloseEvent
from PySide6.QtWidgets import (
@@ -40,9 +39,17 @@ from PySide6.QtWidgets import (
from .. import __version__ as app_version
from ..motion import AxisParams, ConnectionSpec, MotionPoller, MotionService, Snapshot
from ..motion.kinematics import PlatformStack
from ..motion.pose_move import PoseMover, plan_pose_move, plan_return_to_initial
from ..motion.service import clamp_bas_task
from ..motion.units import DEFAULT_UNITS
from ..preferences import load_preferences, save_preferences
from ..sensor import BundleError, load_bundle, load_uv_texture, rasterise, texture_path_for
from ..sensor import (
BundleError,
load_bundle,
load_uv_texture,
rasterise,
texture_path_for,
)
from ..sensor.uv_flat import DEFAULT_FLAT_SIZE
from ..sequence_io import (
FILE_SUFFIX,
@@ -145,6 +152,7 @@ class AutoCalMainWindow(QMainWindow):
# Its size and last tab, from the previous session. Applied here rather than
# in ``_build_ui`` because it is stored state, like the window geometry.
self.device_setup.apply_settings(self.prefs.get("device_setup"))
self.finger_panel.set_confirm(self._confirm_moves)
self.retranslate()
# Load the remembered bundle, so the finger debugger comes up working.
@@ -206,10 +214,16 @@ class AutoCalMainWindow(QMainWindow):
# so the two panels always exist — the window wires them to the controller,
# and the checks and tools reach for them here.
self.device_setup = DeviceSetupDialog(
self.spec, str(self.prefs.get("bas_path") or ""), self
self.spec,
str(self.prefs.get("bas_path") or ""),
self.prefs.get("axis_roles"),
self,
)
self.connection_panel = self.device_setup.connection_panel
self.bus_panel = self.device_setup.bus_panel
# The mapping is live: the guards read it from the service, so an edit on the
# page has to reach the service rather than only the preference file.
self.device_setup.roles_page.rolesChanged.connect(self._on_axis_roles_changed)
self.axis_readout = AxisReadoutPanel(
MotionService().axis_count, self._theme, self.rail_content
@@ -331,6 +345,10 @@ class AutoCalMainWindow(QMainWindow):
self.axis_readout.stopAllRequested.connect(self._on_stop_all)
self.axis_readout.clearAllRequested.connect(self._on_clear_all_alarms)
self.axis_readout.enableAllRequested.connect(self._on_enable_all)
self.axis_readout.returnRequested.connect(self._on_return_to_initial)
self.finger_panel.executeRequested.connect(self._on_execute_pose)
self.finger_panel.stopRequested.connect(self._on_stop_all)
self.finger_panel.confirmChanged.connect(self._on_confirm_moves_changed)
self.run_bar.runRequested.connect(self._on_run_requested)
self.run_bar.pauseRequested.connect(self._on_run_pause)
self.run_bar.abortRequested.connect(self._on_run_abort)
@@ -350,6 +368,20 @@ class AutoCalMainWindow(QMainWindow):
"""The brand bar's chip sends no argument; open on the connection tab."""
self.device_setup.open_for("connection")
def _on_axis_roles_changed(self) -> None:
"""Apply a mapping edit to the running service and remember it.
Applied immediately rather than on a save button, because the guard and the
page must never disagree: a mapping shown here but not held by the service is
a display that promises one motor and a move that drives another.
"""
roles = self.device_setup.roles_page.roles()
self.prefs["axis_roles"] = dict(roles)
self.poller.service.set_axis_roles(roles)
problem = self.poller.service.roles_problem()
if problem:
self.finger_panel.set_move_status(f"✗ {problem}")
# ---------------------------------------------------------- finger debug
def load_bundle(self, path: str, *, flat_size: tuple[int, int] = DEFAULT_FLAT_SIZE):
@@ -724,6 +756,12 @@ class AutoCalMainWindow(QMainWindow):
MotionService(
language=self._language,
bas_task=clamp_bas_task(self.prefs.get("bas_task")),
units_per_mm=float(self.prefs.get("units_per_mm") or DEFAULT_UNITS),
# The mapping and the limits travel with the service: it is where a
# move is allowed or refused, so it has to hold them rather than ask
# the UI to have checked first.
axis_roles=self.prefs.get("axis_roles"),
limits=self.prefs.get("safety_limits"),
),
axes=range(self.axis_panel.axis_count),
)
@@ -733,6 +771,14 @@ class AutoCalMainWindow(QMainWindow):
self.poller.set_interval_ms(int(self.prefs.get("state_poll_ms") or 100))
self.status_strip.set_poll_interval_ms(int(self.prefs.get("state_poll_ms") or 100))
# The pose mover: one axis at a time, advanced by the snapshots this poller
# already produces. Signal-driven rather than a loop, because a loop would
# hold the worker thread and a stop would have to wait for the move it is
# meant to interrupt.
self.mover = PoseMover(self.poller.service, self.poller.submit, self)
self.mover.failed.connect(self._on_move_failed)
self.mover.finished.connect(self._on_move_finished)
# Ages the poll-health chip even when no snapshot arrives. Runs on the
# GUI thread and only reads a timestamp — the controller is untouched —
# so a stopped poll becomes visible instead of leaving stale chips
@@ -783,6 +829,17 @@ class AutoCalMainWindow(QMainWindow):
self.bus_data_panel.set_bus_type(snapshot.bus.bus_type)
self.axis_panel.update_snapshot(snapshot)
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
# state it is deciding from.
if self.mover.running:
self.mover.on_snapshot(snapshot)
self.finger_panel.set_move_status(self.mover.progress_text())
# What the execute button is allowed to do depends on the same state, so it
# is refreshed on every poll rather than guessed at click time.
self.finger_panel.set_move_available(
self._move_available(), moving=self.mover.running
)
# The strip is outside the tabs, so it is fed on every poll regardless
# of which tab is showing.
self.status_strip.set_snapshot(snapshot)
@@ -796,6 +853,22 @@ class AutoCalMainWindow(QMainWindow):
if self.diagnostics_box.isVisible():
self._refresh_diagnostics()
def _move_available(self) -> bool:
"""Whether the platform may be commanded to a pose right now.
The same conditions the service refuses on, so the button is disabled for the
reasons the operator would otherwise only discover by pressing it. The
service still checks — this is the UI telling the truth, not the guard.
"""
snapshot = self._last_snapshot
if not snapshot.connected or not snapshot.bus.initialized:
return False
if self.poller.service.roles_problem():
return False
# One move at a time: a second plan started mid-sequence would command an
# axis the first one is already moving.
return not self.mover.running
def _refresh_status_bar(self) -> None:
snapshot = self._last_snapshot
if not snapshot.connected:
@@ -947,8 +1020,113 @@ class AutoCalMainWindow(QMainWindow):
def _stop_all(service: MotionService) -> object:
return service.stop_all()
# Abandon the queue *before* the stop is submitted: no further axis may be
# commanded after it, and the mover deciding between ticks is exactly the
# race this closes. Stopping the axis that is moving is the controller's job.
if self.mover.running:
self.mover.cancel(t("move.stopped"))
self.finger_panel.set_move_status(t("move.stopped"))
self.poller.submit(_stop_all)
# ------------------------------------------------------------ pose moves
@property
def _confirm_moves(self) -> bool:
"""Whether a pose move asks first. The operator's choice, remembered."""
return bool(self.prefs.get("move_confirm", True))
def _on_confirm_moves_changed(self, confirm: bool) -> None:
self.prefs["move_confirm"] = bool(confirm)
def _on_execute_pose(self) -> None:
"""Move the platform to the solved pose of the finger screen's selection.
Stages and table only: the press axis is deliberately not commanded, so this
is workflow P6's dry run for one point — it cannot reach the sensor. The
numbers shown in the confirmation are the ones the preview drew, so the
operator checks the simulation against the machine, not a summary of it.
"""
solve = self.finger_panel.solved_pose()
if solve is None:
self._refuse_move(t("move.no_point"))
return
targets, held = solve
plan = plan_pose_move(
self.poller.service,
targets,
title=t("move.execute"),
hold=held,
)
self._start_move(plan, t("move.execute"))
def _on_return_to_initial(self) -> None:
"""Every drive back to zero, press axis first."""
plan = plan_return_to_initial(self.poller.service)
self._start_move(plan, t("move.return_initial"))
def _start_move(self, plan, title: str) -> None:
service = self.poller.service
refusal = service.check_axes_move(
{step.role: step.target for step in plan.steps}
)
if refusal is not None:
self._refuse_move(refusal.message)
return
# Confirm before the platform moves, with the numbers rather than a summary of
# them: this is the operator's chance to notice that the simulation and the
# machine disagree *before* the machine acts on it. Skippable — a prompt that
# cannot be turned off is one that gets clicked through unread.
if self._confirm_moves:
steps = "\n".join(
f" {index + 1}. {step.role} → {step.target:+.4f} {step.unit}"
f" (驱动器 {step.axis})"
for index, step in enumerate(plan.steps)
)
if plan.held:
steps += "\n\n " + t("move.hold_note", roles=", ".join(plan.held))
body_key = (
"move.confirm_return"
if title == t("move.return_initial")
else "move.confirm_execute"
)
answer = QMessageBox.question(
self,
t("move.confirm_title"),
t(body_key, steps=steps),
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if answer != QMessageBox.StandardButton.Yes:
self.finger_panel.set_move_status(t("move.cancelled"))
return
if not self.mover.start(plan).ok:
self._refuse_move(self.mover.state.failure or t("move.busy"))
return
self.finger_panel.set_move_status(plan.describe())
self.status_strip.set_action_result(f"{title}: {plan.describe()}", True)
def _refuse_move(self, reason: str) -> None:
"""A refused move is reported where the operator asked for it."""
self._last_error = reason
self.finger_panel.set_move_status(f"✗ {reason}")
text = f"{t('move.refused')}: {reason}"
self.statusBar().showMessage(text)
self.status_strip.set_action_result(text, False)
logger.warning("Move refused: %s", reason)
def _on_move_finished(self, plan) -> None:
if plan is not None:
self.finger_panel.set_move_status(f"✓ {plan.title or t('move.done')}")
self.status_strip.set_action_result(t("move.done"), True)
def _on_move_failed(self, message: str) -> None:
# A mover that gave up leaves an axis wherever it stopped: stop the rest
# rather than let the next step continue a sequence nobody has verified.
self.finger_panel.set_move_status(f"✗ {message}")
self.status_strip.set_action_result(f"{t('move.failed')}: {message}", False)
self.statusBar().showMessage(f"{t('move.failed')}: {message}")
self.poller.submit(lambda service: service.stop_all())
def _on_move_abs(self, axis: int, target: float) -> None:
self.poller.submit(lambda service: service.move_abs(axis, target))
@@ -12,6 +12,7 @@ will fill it, so nobody mistakes an empty page for a broken one.
from __future__ import annotations
from orisys_app_shared.i18n import t
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QLabel,
@@ -20,8 +21,6 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_app_shared.i18n import t
from .axis_panel import AxisPanel
from .sequence_panel import SequencePanel