diff --git a/apps/autocal/README.md b/apps/autocal/README.md index f8d2365..1fd465f 100644 --- a/apps/autocal/README.md +++ b/apps/autocal/README.md @@ -57,6 +57,54 @@ the wrong runtime version is the kind of fault that looks like a working machine until it moves, so the override is authoritative: if it does not contain `zauxdll.dll`, the app refuses to connect instead of quietly falling back. +One contract in `shared/src/orisys_app_shared/zmotion/client.py` is worth +knowing before touching it: `ZAux_Execute` **must** be given a reply buffer. +Passing `NULL` with a zero length is a write through a null pointer inside the +DLL — a native crash the worker thread cannot catch — and the buffer is also +how a refused online command explains itself, because `ZAux_Execute` returns 0 +for a `RUN` whose file the controller does not have. + +## Bring-up sequence + +The vendor sequence, with the checks AutoCal's own state machine adds. P1 of +`docs/workflow.md` is this list in the workflow's vocabulary. + +1. **Connect** (`eth` + the controller's IP). On success the app adopts any BAS + program the controller is already running from a previous session instead of + claiming the bus is uninitialised; `BUS_TYPE` is read immediately, and + `Bus_InitStatus` on the first poll. +2. **Download only if the bus type is unknown.** `ECAT.bas` / `RTEX.bas` are + chosen from `config/bas/` (the vendor requires a non-Chinese path) and + written to ROM. A controller that still has the program needs no download. +3. **Wait for `Bus_InitStatus`** — `−1` running, `1` success, `0` failed — + inside the vendor's 25 s window, then a timeout message naming the ZDevelop + diagnostic (`?*ETHERCAT`). +4. **Re-initialise** re-runs the program without re-downloading it: + `STOPTASK ` then `RUN "",`. The task number is + `bas_task` in `preference.json` (default **4**, the task the vendor + reference uses); the file name comes from this session's download, or from the + remembered `bas_path` when the session adopted a running program (the + controller does not report the name). The controller answers a `RUN` for a + file it does not have with **code 0** and the refusal in its reply text, so + that text is checked and shown rather than reported as an initialisation in + progress. +5. **Enable, then jog.** Motion stays refused until step 3 reports success. + +### Connection and error codes + +`ZAux_*` returns two unrelated families of code, and they need different answers: + +| Range | Family | What the operator should do | +|---|---|---| +| `0` | success | — | +| `20000–29999` | PC-library fault (handshake, socket, address) | check the cable/segment and the IP, make sure nothing else holds the connection, then retry. Frequently transient: a first open on a cold route can fail and succeed on the next attempt, which is why an Ethernet connect makes up to 3 attempts before reporting. | +| `30000+` | the auxiliary block `zauxdll2.h` documents | the message names it (`无应答`, `参数错误`, …) | +| anything else | the controller's own code | look it up in ZDevelop's table | + +A **host-library** code is never sent to ZDevelop's table, because that table +cannot contain it. Observed at the bench: `20008` from `ZAux_OpenEth` on a first, +transient attempt that succeeded immediately afterwards. + ## Safety model The guards live in `motion/service.py`, not in the UI, so they hold whatever diff --git a/apps/autocal/config/preferences.default.json b/apps/autocal/config/preferences.default.json index 6dec09f..4caa949 100644 --- a/apps/autocal/config/preferences.default.json +++ b/apps/autocal/config/preferences.default.json @@ -10,6 +10,7 @@ "local_string": "0" }, "bas_path": "", + "bas_task": 4, "units_per_mm": 10.0, "state_poll_ms": 100, "sdo_last": { "node": 0, "index": 24704, "subindex": 0, "type": 0 }, diff --git a/apps/autocal/docs/workflow.md b/apps/autocal/docs/workflow.md index ff2bdeb..d86c5c8 100644 --- a/apps/autocal/docs/workflow.md +++ b/apps/autocal/docs/workflow.md @@ -61,17 +61,38 @@ slaves. A missing drive discovered at P7 wastes the whole run. **Inputs:** controller address, the matching `.bas` program (`ECAT.bas` / `RTEX.bas`). -**Steps:** connect → download and run the BAS program → watch -`Bus_InitStatus` to completion or timeout → read `BUS_TYPE`, -`Bus_TotalAxisnum`, and the node count. +**Steps:** connect → **adopt a program the controller is already running, or +download one if `BUS_TYPE` is unreadable** → watch `Bus_InitStatus` to completion +or timeout → read `BUS_TYPE`, `Bus_TotalAxisnum`, and the node count. Step 2 is +conditional because a controller usually still has the program in ROM from the +last session, and a download is a bus re-scan the operator did not ask for. + +**Re-initialising without a download:** `STOPTASK ` then +`RUN "",`, on the task from `bas_task` in `preference.json` +(default 4 — the task the vendor reference uses; stopping another task restarts +nothing). Two things are easy to get wrong here, and both have been: + +- the file name must be the one the *controller* has. When the session adopted a + running program, the controller does not report the name, so the app carries + the remembered `bas_path` into the connect for exactly this purpose; +- the controller answers a `RUN` naming a file it does not have with **code 0** + and the diagnosis in its reply text. The reply is checked, and its text is + shown instead of reporting an initialisation that is not happening. **Outputs:** bus type, node count, axis count, init status. **Gate:** init status is success, **and** the node/axis counts match what this bench is expected to have. A mismatch is a finding, not a warning. -**Existing code:** `MotionService.download_bas` / `poll` already implement the -state machine and the 25 s timeout. +**Existing code:** `MotionService.download_bas` / `reinit_bus` / `poll` implement +the state machine and the 25 s timeout. + +**Connection faults are a different family.** A `ZAux_*` code in 20000–29999 is a +PC-library fault (handshake, socket, address), not a controller code: ZDevelop's +table cannot contain it, so the app answers it with cable/IP/no-second-client +advice and retries an Ethernet open up to three times, because a first open on a +cold route can fail and succeed immediately after. `20008` at the bench was that +case. --- diff --git a/apps/autocal/src/orisys_autocal/motion/__init__.py b/apps/autocal/src/orisys_autocal/motion/__init__.py index 749e6fe..1af5a18 100644 --- a/apps/autocal/src/orisys_autocal/motion/__init__.py +++ b/apps/autocal/src/orisys_autocal/motion/__init__.py @@ -16,20 +16,25 @@ from .models import ( Snapshot, ) from .poller import MotionPoller -from .service import PLATFORM_AXES, MotionService +from .service import DEFAULT_BAS_TASK, PLATFORM_AXES, MotionService, clamp_bas_task from .transport import DemoTransport, NativeTransport, ZmcTransport from .units import ( DEFAULT_ACCEL, DEFAULT_SPEED, DEFAULT_UNITS, PULSES_PER_MM, + ROTARY_UNITS_PER_90_DEG, + deg_to_user_units, mm_to_user_units, safe_float, + units_per_degree, + user_units_to_deg, user_units_to_mm, ) __all__ = [ "DEFAULT_ACCEL", + "DEFAULT_BAS_TASK", "DEFAULT_SPEED", "DEFAULT_UNITS", "INIT_FAILED", @@ -38,6 +43,7 @@ __all__ = [ "INIT_TIMEOUT_S", "PLATFORM_AXES", "PULSES_PER_MM", + "ROTARY_UNITS_PER_90_DEG", "AxisParams", "AxisState", "AxisStatusBits", @@ -52,7 +58,11 @@ __all__ = [ "NativeTransport", "Snapshot", "ZmcTransport", + "clamp_bas_task", + "deg_to_user_units", "mm_to_user_units", "safe_float", + "units_per_degree", + "user_units_to_deg", "user_units_to_mm", ] diff --git a/apps/autocal/src/orisys_autocal/motion/errors.py b/apps/autocal/src/orisys_autocal/motion/errors.py index b7cd5c4..41ca39e 100644 --- a/apps/autocal/src/orisys_autocal/motion/errors.py +++ b/apps/autocal/src/orisys_autocal/motion/errors.py @@ -26,6 +26,14 @@ logger = logging.getLogger(__name__) #: Auxiliary (host-side) error block offset, from ``zauxdll2.h``. ERR_AUX_OFFSET = 30000 +#: The block the PC library uses for its *own* faults — handshake, socket, +#: address, a controller that stopped answering mid-session. It sits below +#: ``ERR_AUX_OFFSET``, and it is **not** in the firmware's code table: pointing +#: an operator at ZDevelop for one of these sends them to a table that cannot +#: contain the number. Observed live: ``20008`` from ``ZAux_OpenEth`` on a first, +#: transient attempt that succeeded on the next. +HOST_LIB_ERROR_RANGE = (20000, 29999) + #: Meaning per code, as ``(english, chinese)``. MESSAGES: dict[int, tuple[str, str]] = { ERR_AUX_OFFSET: ("No response from the controller", "控制器无应答"), @@ -42,6 +50,24 @@ UNKNOWN_HINT = ( "or run the same command from ZDevelop to see the decoded fault." ) +#: Advice for a host-library fault, which is a communication problem the +#: operator can actually act on — and which is frequently transient. +HOST_LIB_HINT_ZH = ( + "上位机库通信错误:请确认网线/网段与控制器 IP," + "确认控制器已上电、没有其它程序占用连接,然后重试。" +) +HOST_LIB_HINT_EN = ( + "Host-library communication error: check the cable/segment and the " + "controller IP, make sure the controller is powered and no other program " + "holds the connection, then retry." +) + + +def is_host_lib_error(code: int) -> bool: + """True for a PC-library fault, which is not a controller code.""" + low, high = HOST_LIB_ERROR_RANGE + return low <= int(code) <= high + def describe(code: int, *, language: str = "zh_CN") -> str: """A human-readable message for a controller return code. @@ -60,18 +86,23 @@ def describe(code: int, *, language: str = "zh_CN") -> str: index = 1 if language == "zh_CN" else 0 known = MESSAGES.get(code) - if known is None: - # An unknown code that still resolves to an auxiliary slot is more - # useful than a bare integer, but only when it is documented. - logger.debug("Undocumented ZMotion return code %s", code) - prefix = "错误码" if language == "zh_CN" else "Error code" + if known is not None: + return known[index] + + logger.debug("Undocumented ZMotion return code %s", code) + if is_host_lib_error(code): + prefix = "上位机库错误" if language == "zh_CN" else "Host-library error" return f"{prefix} {code}" - return known[index] + prefix = "错误码" if language == "zh_CN" else "Error code" + return f"{prefix} {code}" def describe_with_hint(code: int, *, language: str = "zh_CN") -> str: - """As :func:`describe`, plus the ZDevelop pointer for unknown codes.""" + """As :func:`describe`, plus the advice the code's family warrants.""" message = describe(code, language=language) if int(code) == 0 or int(code) in MESSAGES: return message + if is_host_lib_error(code): + hint = HOST_LIB_HINT_ZH if language == "zh_CN" else HOST_LIB_HINT_EN + return f"{message} — {hint}" return f"{message} — {UNKNOWN_HINT}" diff --git a/apps/autocal/src/orisys_autocal/motion/models.py b/apps/autocal/src/orisys_autocal/motion/models.py index 3fa1806..5a33c5a 100644 --- a/apps/autocal/src/orisys_autocal/motion/models.py +++ b/apps/autocal/src/orisys_autocal/motion/models.py @@ -134,11 +134,17 @@ class BusStatus: def supports_rtex(self) -> bool: return self.bus_type == "rtex" - def timed_out(self, now: float) -> bool: - """True while a BAS run started long ago has still not reported.""" + def timed_out(self, now: float, timeout: float = INIT_TIMEOUT_S) -> bool: + """True when a BAS run started long ago has still not reported. + + ``timeout`` is the window to judge against. It defaults to the vendor's + 25 s, but a caller that owns a configurable window (the service) must + pass it: reading the module constant here instead made the configured + timeout invisible to every caller that asked this question. + """ if self.init_status != INIT_RUNNING or self.init_started_at is None: return False - return (now - self.init_started_at) > INIT_TIMEOUT_S + return (now - self.init_started_at) > float(timeout) @dataclass(frozen=True) diff --git a/apps/autocal/src/orisys_autocal/motion/poller.py b/apps/autocal/src/orisys_autocal/motion/poller.py index 4ae065e..1791a55 100644 --- a/apps/autocal/src/orisys_autocal/motion/poller.py +++ b/apps/autocal/src/orisys_autocal/motion/poller.py @@ -279,8 +279,8 @@ class MotionPoller(QObject): """Queue a service call. Returns the id it will be reported under.""" return self._worker.submit(action) - def connect_controller(self, spec) -> int: - return self.submit(lambda service: service.connect(spec)) + def connect_controller(self, spec, *, bas_path: str = "") -> int: + return self.submit(lambda service: service.connect(spec, bas_path=bas_path)) def disconnect_controller(self) -> int: def _disconnect(service: MotionService): diff --git a/apps/autocal/src/orisys_autocal/motion/service.py b/apps/autocal/src/orisys_autocal/motion/service.py index 8379c62..f1fb879 100644 --- a/apps/autocal/src/orisys_autocal/motion/service.py +++ b/apps/autocal/src/orisys_autocal/motion/service.py @@ -52,6 +52,48 @@ PLATFORM_AXES = 5 #: ``BUS_TYPE`` values written by the vendor ``.bas`` programs. _BUS_TYPE_NAMES: dict[int, BusType] = {0: "ecat", 1: "rtex"} +#: The controller task the vendor reference starts its BAS program on +#: (``RUN "",4``). Re-initialisation must name the *same* task, or it +#: stops a task that was never running and restarts nothing. +DEFAULT_BAS_TASK = 4 + +#: Accepted task range. The controller's task numbers are small; anything +#: outside this is a typed-in mistake, not a task to send. +_BAS_TASK_RANGE = (1, 8) + +#: Substrings the firmware puts in an online command's reply when the command +#: was refused *without* a non-zero return code — a ``RUN`` naming a program the +#: controller does not have is the one that matters here (``ZAux_Execute`` +#: returns 0 for it). +_ONLINE_FAILURE_MARKERS = ("not find", "fail", "error") + + +def clamp_bas_task(value: object, fallback: int = DEFAULT_BAS_TASK) -> int: + """A usable controller task number, or the vendor default.""" + try: + task = int(value) + except (TypeError, ValueError): + return fallback + low, high = _BAS_TASK_RANGE + return task if low <= task <= high else fallback + + +def _online_failure(reply: str) -> str: + """The controller's own refusal text from an online command, or ``""``. + + ``ZAux_Execute`` reports a rejected online command in its reply rather than + in its return code, so a caller that only checks the code cannot tell + "started" from "no such program". Anything the firmware answers that carries + a failure marker is returned for the UI to show verbatim. + """ + text = " ".join(str(reply or "").split()) + if not text: + return "" + lowered = text.lower() + if any(marker in lowered for marker in _ONLINE_FAILURE_MARKERS): + return text + return "" + class MotionService: """The controller session: one connection, five axes, one state machine.""" @@ -64,12 +106,14 @@ class MotionService: units_per_mm: float = DEFAULT_UNITS, pulses_per_mm: float = PULSES_PER_MM, language: str = "zh_CN", + bas_task: int = DEFAULT_BAS_TASK, ) -> None: self._transport: ZmcTransport = transport or NativeTransport() self.axis_count = int(axis_count) self._units_per_mm = safe_float(units_per_mm, DEFAULT_UNITS) self._pulses_per_mm = pulses_per_mm self.language = language + self.bas_task = clamp_bas_task(bas_task) self._spec: ConnectionSpec | None = None self._bus = BusStatus() @@ -125,11 +169,18 @@ class MotionService: # ------------------------------------------------------------ connect/init - def connect(self, spec: ConnectionSpec) -> MotionResult: + def connect(self, spec: ConnectionSpec, *, bas_path: str = "") -> MotionResult: """Open the controller. A previous connection is closed first: switching transport type without closing leaks the DLL handle, which the reference program does. + + ``bas_path`` is the ``.bas`` file the operator last used, as remembered + by the UI. It is *recorded*, never downloaded: a controller usually still + has the program in ROM from the last session, and re-writing it on every + connect is both slow and a bus re-scan the operator did not ask for. Its + only job is to make :meth:`reinit_bus` possible on a session that adopted + a running program instead of downloading one. """ if self._spec is not None: self.disconnect() @@ -146,25 +197,35 @@ class MotionService: logger.info("Connected: %s via %s", spec.describe(), self._transport.source) # A controller may already have a BAS program running from a previous # session; adopt it rather than claiming the bus is uninitialised. - self._adopt_existing_bas() + self._adopt_existing_bas(bas_path=bas_path) return MotionResult.success(f"已连接 {spec.describe()}" if self._is_zh else f"Connected {spec.describe()}") @property def _is_zh(self) -> bool: return self.language == "zh_CN" - def _adopt_existing_bas(self) -> None: + def _adopt_existing_bas(self, *, bas_path: str = "") -> None: try: bus_type = self._transport.user_var("BUS_TYPE") except (ZAuxError, OSError): self._bas_loaded = False self._bus = BusStatus(bus_type="unknown", init_status=INIT_RUNNING) return + path = str(bas_path or "").strip() self._bas_loaded = True + # ``init_status`` stays "running" until the first poll reads the real + # ``Bus_InitStatus``: connect() must not claim a state it has not read. + # The adoption is *timestamped* though, so a controller that never + # answers still reaches the timeout instead of looking busy forever — + # and so "re-initialise" is not refused as "already running" five + # minutes into a session. + self._init_started_at = time.monotonic() self._bus = BusStatus( bus_type=_BUS_TYPE_NAMES.get(int(bus_type), "unknown"), init_status=INIT_RUNNING, + bas_path=path, bas_loaded=True, + init_started_at=self._init_started_at, ) def disconnect(self) -> None: @@ -224,29 +285,50 @@ class MotionService: ) def reinit_bus(self) -> MotionResult: - """Re-run the BAS program's init task without re-downloading it.""" + """Re-run the BAS program's init task without re-downloading it. + + The vendor sequence, exactly: stop the task the program runs on, then + ``RUN`` the file on that same task. Task numbers are not interchangeable + — the reference program starts on task 4, so stopping task 2 restarts + nothing. + + ``ZAux_Execute`` returns 0 for a ``RUN`` naming a program the controller + does not have, putting the diagnosis in the reply text instead. That text + is checked here rather than trusted: a re-initialisation that cannot + happen must not be reported as one that is running. + """ if not self.connected: return MotionResult.refused(self._text("未连接控制器", "Not connected")) if not self._bas_loaded or not self._bus.bas_path: return MotionResult.refused( self._text( - "Bas文件未加载,请先下载 ECAT.bas 或 RTEX.bas", - "No BAS program loaded; download ECAT.bas or RTEX.bas first", + "没有 BASIC 程序文件名,无法重新初始化;请先下载 ECAT.bas 或 RTEX.bas", + "No BAS program file name is known, so it cannot be " + "re-initialised; download ECAT.bas or RTEX.bas first", ) ) - if self._bus.init_status == INIT_RUNNING and not self._bus.timed_out(time.monotonic()): + if self._bus.init_status == INIT_RUNNING and not self._bus.timed_out( + time.monotonic(), self._init_timeout + ): return MotionResult.refused( self._text("总线正在初始化,请稍候", "Bus initialisation is already running") ) script = os.path.basename(self._bus.bas_path) + task = clamp_bas_task(self.bas_task) try: - self._transport.execute("STOPTASK 2") - self._transport.execute(f'RUN "{script}",4') + stop_reply = self._transport.execute(f"STOPTASK {task}") + run_reply = self._transport.execute(f'RUN "{script}",{task}') except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) return MotionResult.failed(code, self._msg(code) if code else str(exc)) + # The stop command has no meaningful reply; the RUN one does. + failure = _online_failure(run_reply) + if failure: + logger.warning("Re-initialise rejected for %s on task %s: %s", script, task, failure) + return MotionResult.failed(0, failure) + self._init_started_at = time.monotonic() self._bus = BusStatus( bus_type=self._bus.bus_type, @@ -255,6 +337,19 @@ class MotionService: bas_loaded=True, init_started_at=self._init_started_at, ) + # Re-read BUS_TYPE now: the program may be a different family than the + # one the previous session left behind. + try: + bus_type = self._transport.user_var("BUS_TYPE") + self._bus = _replace_bus_type(self._bus, _BUS_TYPE_NAMES.get(int(bus_type), "unknown")) + except (ZAuxError, OSError): + pass + logger.info( + "Re-initialising %s on controller task %s (stop reply=%r)", + script, + task, + stop_reply, + ) return MotionResult.success( self._text("正在重新初始化总线", "Re-initialising the bus") ) diff --git a/apps/autocal/src/orisys_autocal/motion/transport.py b/apps/autocal/src/orisys_autocal/motion/transport.py index 25d7dbc..d3e189f 100644 --- a/apps/autocal/src/orisys_autocal/motion/transport.py +++ b/apps/autocal/src/orisys_autocal/motion/transport.py @@ -17,7 +17,9 @@ Three implementations: from __future__ import annotations +import logging import math +import os import time from typing import Protocol, runtime_checkable @@ -26,6 +28,8 @@ from orisys_app_shared.zmotion import ZAuxError, ZmotionClient from .errors import ERR_AUX_OFFSET from .models import BusType, ConnectionSpec +logger = logging.getLogger(__name__) + @runtime_checkable class ZmcTransport(Protocol): @@ -42,7 +46,7 @@ class ZmcTransport(Protocol): def bas_download(self, filename: str, run_mode: int = 1) -> None: ... - def execute(self, command: str) -> None: ... + def execute(self, command: str) -> str: ... def user_var(self, name: str) -> float: ... @@ -114,6 +118,14 @@ class ZmcTransport(Protocol): class NativeTransport: """The real controller, through the vendored ZMotion runtime.""" + #: How many times an Ethernet open is attempted before the failure is + #: reported, and how long to wait between attempts. A first connection to a + #: controller on a cold ARP/route table can fail once and succeed on the + #: next try — the reference program's answer is "click connect again", which + #: the operator should not have to know. + ETH_OPEN_ATTEMPTS = 3 + ETH_OPEN_RETRY_DELAY_S = 0.3 + def __init__(self, client: ZmotionClient | None = None) -> None: self._client = client or ZmotionClient() @@ -127,16 +139,39 @@ class NativeTransport: def open(self, spec: ConnectionSpec) -> None: if spec.mode == "eth": - result = self._client.open_eth(spec.ip) + self._open_eth_with_retry(spec.ip) elif spec.mode == "com": result = self._client.open_com(spec.com_id) + result.raise_for_status(f"open {spec.describe()}") elif spec.mode == "pci": result = self._client.open_pci(spec.card_num) + result.raise_for_status(f"open {spec.describe()}") elif spec.mode == "local": result = self._client.fast_open(5, spec.local_string, 1000) + result.raise_for_status(f"open {spec.describe()}") else: raise ValueError(f"{spec.mode!r} is not a native connection mode") - result.raise_for_status(f"open {spec.describe()}") + + def _open_eth_with_retry(self, ip: str) -> None: + """Open over Ethernet, retrying a *failed* attempt a couple of times.""" + last_code = 0 + for attempt in range(1, self.ETH_OPEN_ATTEMPTS + 1): + result = self._client.open_eth(ip) + if result.ok: + if attempt > 1: + logger.info("Ethernet open of %s succeeded on attempt %s", ip, attempt) + return + last_code = result.code + logger.warning( + "Ethernet open of %s failed on attempt %s/%s with code %s", + ip, + attempt, + self.ETH_OPEN_ATTEMPTS, + result.code, + ) + if attempt < self.ETH_OPEN_ATTEMPTS: + time.sleep(self.ETH_OPEN_RETRY_DELAY_S) + ZAuxError(last_code, f"open {ip}").raise_for_status(f"open {ip}") def close(self) -> None: self._client.close() @@ -152,8 +187,15 @@ class NativeTransport: def bas_download(self, filename: str, run_mode: int = 1) -> None: self._client.bas_download(filename, run_mode).raise_for_status("BAS download") - def execute(self, command: str) -> None: - self._client.execute(command).raise_for_status(command) + def execute(self, command: str) -> str: + """Run an online command; return the controller's own reply text. + + The reply is not decoration: an online command the firmware refuses + (``RUN`` naming a program it does not have) returns code 0 and explains + itself only in this text. See + :meth:`orisys_autocal.motion.service.MotionService.reinit_bus`. + """ + return str(self._client.execute(command).raise_for_status(command).value or "") def user_var(self, name: str) -> float: return float(self._client.get_user_var(name).raise_for_status(name).value) @@ -278,6 +320,7 @@ class DemoTransport: self._homing_until = {axis: 0.0 for axis in range(self._axes)} self._home_done = {axis: False for axis in range(self._axes)} self._bas_loaded = False + self._bas_file = "" self._init_started: float | None = None self._opened = False self._sdo: dict[tuple[int, int, int], int] = {} @@ -301,15 +344,25 @@ class DemoTransport: if not self._opened: raise ZAuxError(ERR_NOT_OPEN, "not connected") self._bas_loaded = True + # The simulation keeps the file name, so a re-init can tell a program + # the controller has from one it does not — the distinction the real + # firmware makes in an online command's reply text. + self._bas_file = os.path.basename(str(filename).replace("\\", "/")) # ECAT.bas delays 3 s before scanning; mimic the delay so the UI's # "initialising" state is visible rather than instantaneous. self._init_started = time.monotonic() + DEMO_INIT_DELAY_S - def execute(self, command: str) -> None: + def execute(self, command: str) -> str: if not self._opened: raise ZAuxError(ERR_NOT_OPEN, "not connected") if "RUN" in command.upper() and self._bas_loaded: + wanted = _run_filename(command) + if wanted and self._bas_file and wanted != self._bas_file: + # The real firmware returns code 0 here and explains itself only + # in this text. See MotionService.reinit_bus. + return f"Online command error, Run filename:{wanted} not find." self._init_started = time.monotonic() + DEMO_INIT_DELAY_S + return "" def user_var(self, name: str) -> float: if not self._bas_loaded: @@ -465,3 +518,19 @@ DEMO_HOME_S = 2.0 #: Demo-only failure codes, kept outside the documented vendor range on purpose. ERR_NOT_OPEN = 30500 ERR_DEMO_UNSUPPORTED = 30501 + + +def _run_filename(command: str) -> str: + """The quoted file name in an online ``RUN "file",task`` command, or ``""``. + + Shared by the simulated controller and its tests so the demo rejects the + same commands the real firmware rejects. + """ + text = str(command or "") + start = text.find('"') + if start < 0: + return "" + end = text.find('"', start + 1) + if end < 0: + return "" + return text[start + 1 : end].strip() diff --git a/apps/autocal/src/orisys_autocal/motion/units.py b/apps/autocal/src/orisys_autocal/motion/units.py index 0cd7f77..02addac 100644 --- a/apps/autocal/src/orisys_autocal/motion/units.py +++ b/apps/autocal/src/orisys_autocal/motion/units.py @@ -1,31 +1,39 @@ -"""User-unit to millimetre conversion. +"""Platform unit conversion (linear mm and rotary degrees). -The numbers come from bench measurement on the platform, transcribed from the -vendor reference program's comments:: +Bench measurements (controller user units, with the platform's usual +``UNITS = 10`` pulse equivalent unless noted):: - # 轴0(X)/轴1(Y) 实测: UNITS=10 时 1000用户单位 = 2mm → 每毫米 5000 脉冲 + # 轴0(X)/轴1(Y): UNITS=10 时 1000用户单位 = 2mm → 每毫米 5000 脉冲 # 当量5 → 1000单位=1mm,不是100单位=1mm + # 轴2: 12500 用户单位 = 90° + # 轴3: 29500 用户单位 = 90° -so ``PULSES_PER_MM = 5000`` is the calibration constant and ``UNITS = 10`` is -the pulse equivalent the platform is configured with. The displayed millimetre -value is therefore *derived*, not measured: the UI labels it as such, and no -code path may treat it as a calibrated length. +Linear display millimetres use ``PULSES_PER_MM`` and the axis pulse equivalent. +Rotary display degrees use :data:`ROTARY_UNITS_PER_90_DEG` for axes 2 and 3. +Displayed values are *derived*, not fresh measurements. """ from __future__ import annotations import math -#: Measured: 5000 pulses per millimetre (see the module docstring). +#: Measured: 5000 pulses per millimetre (axes 0 / 1). PULSES_PER_MM = 5000.0 -#: Pulse equivalent the platform is configured with by default. +#: Pulse equivalent the linear axes are configured with by default. DEFAULT_UNITS = 10.0 #: Move-parameter defaults, matching the reference program's UI initialisation. DEFAULT_SPEED = 1000.0 DEFAULT_ACCEL = 10000.0 +#: Bench: user units that correspond to a 90° turn, keyed by axis number. +#: Axis 2 and 3 are rotary; axes 0 / 1 / 4 stay out of this map until measured. +ROTARY_UNITS_PER_90_DEG: dict[int, float] = { + 2: 12500.0, + 3: 29500.0, +} + def safe_float(text: object, fallback: float) -> float: """Parse ``text`` as a finite float, falling back on anything unusable. @@ -70,3 +78,27 @@ def mm_to_user_units( if units <= 0: return math.nan return float(value) * float(pulses_per_mm) / float(units) + + +def units_per_degree(axis: int) -> float: + """User units per degree for a rotary axis, or NaN if not calibrated.""" + per_90 = ROTARY_UNITS_PER_90_DEG.get(int(axis)) + if per_90 is None or per_90 <= 0: + return math.nan + return float(per_90) / 90.0 + + +def user_units_to_deg(axis: int, value: float) -> float: + """Convert controller user units to degrees for a calibrated rotary axis.""" + scale = units_per_degree(axis) + if not math.isfinite(scale) or scale == 0.0: + return math.nan + return float(value) / scale + + +def deg_to_user_units(axis: int, degrees: float) -> float: + """Inverse of :func:`user_units_to_deg`.""" + scale = units_per_degree(axis) + if not math.isfinite(scale): + return math.nan + return float(degrees) * scale diff --git a/apps/autocal/src/orisys_autocal/preferences.py b/apps/autocal/src/orisys_autocal/preferences.py index 2a7447e..c6a928d 100644 --- a/apps/autocal/src/orisys_autocal/preferences.py +++ b/apps/autocal/src/orisys_autocal/preferences.py @@ -13,7 +13,7 @@ from typing import Any from orisys_app_shared import prefs as _prefs from orisys_app_shared.paths import preference_path -from .motion.service import PLATFORM_AXES +from .motion.service import 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. @@ -155,6 +155,11 @@ def default_preferences() -> dict[str, Any]: # Remembered so "download the BAS program" is one click after the first # time, instead of re-picking the file every session. "bas_path": "", + #: The controller task the BAS program runs on. The vendor reference + #: starts it on 4 and re-initialises it with ``RUN "",4``; stopping + #: any other task restarts nothing. Kept configurable because the task + #: number is a property of the project loaded on the controller. + "bas_task": DEFAULT_BAS_TASK, #: Displayed millimetres are derived from this; it is a bench-measured #: constant, not a calibrated length (see motion/units.py). "units_per_mm": DEFAULT_UNITS, @@ -245,6 +250,8 @@ def load_preferences() -> dict[str, Any]: if value and not tool[index]: tool[index] = value 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")) return prefs diff --git a/apps/autocal/src/orisys_autocal/ui/connection_panel.py b/apps/autocal/src/orisys_autocal/ui/connection_panel.py index 31d4e3e..a76ec05 100644 --- a/apps/autocal/src/orisys_autocal/ui/connection_panel.py +++ b/apps/autocal/src/orisys_autocal/ui/connection_panel.py @@ -146,13 +146,17 @@ class ConnectionPanel(QGroupBox): self.address_edit.setText("") def set_ips(self, ips: list[str]) -> None: - """Offer scanned addresses for the Ethernet mode.""" + """Offer scanned addresses for the Ethernet mode. + + The hint is shown whatever the address box currently holds: a scan that + found the controller is exactly the moment the operator wants to see the + address, and suppressing it because the box already held a different one + made the button look broken. The box itself is never overwritten, so a + typed address survives a scan. + """ if self.mode_combo.currentData() != "eth" or not ips: return - current = self.address_edit.text().strip() - if current and current not in ips: - return - self.ip_hint.setText(" ".join(ips[:4])) + self.ip_hint.setText(" ".join(str(ip) for ip in ips[:4])) def set_connected(self, connected: bool, source: str = "") -> None: self.open_button.setEnabled(not connected) diff --git a/apps/autocal/src/orisys_autocal/ui/window.py b/apps/autocal/src/orisys_autocal/ui/window.py index c5b6875..be9a6ab 100644 --- a/apps/autocal/src/orisys_autocal/ui/window.py +++ b/apps/autocal/src/orisys_autocal/ui/window.py @@ -40,6 +40,7 @@ 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.service import clamp_bas_task from ..preferences import load_preferences, save_preferences from ..sensor import BundleError, load_bundle, load_uv_texture, rasterise, texture_path_for from ..sensor.uv_flat import DEFAULT_FLAT_SIZE @@ -684,7 +685,10 @@ class AutoCalMainWindow(QMainWindow): def _build_poller(self) -> None: self.poller = MotionPoller( - MotionService(language=self._language), + MotionService( + language=self._language, + bas_task=clamp_bas_task(self.prefs.get("bas_task")), + ), axes=range(self.axis_panel.axis_count), ) self.poller.snapshotReady.connect(self._on_snapshot) @@ -708,7 +712,12 @@ class AutoCalMainWindow(QMainWindow): ) -> int: """Ask the poller to open the controller. Returns the command id.""" target = spec or self.connection_panel.spec - token = self.poller.connect_controller(target) + # The remembered BAS file travels with the connect so that a session + # which adopts a program already running on the controller can still + # re-initialise it (the controller does not report the file name). + token = self.poller.connect_controller( + target, bas_path=str(self.prefs.get("bas_path") or "") + ) self._pending[token] = label self.connection_panel.set_status(t("connection.state_connecting")) return token @@ -799,7 +808,12 @@ class AutoCalMainWindow(QMainWindow): message, ok = _interpret_result(result) if ok: self._last_error = "" - if message: + if label_key == "connection.scan" and isinstance(result, (list, tuple)): + # A scan exists to put the addresses somewhere the operator can + # click; reporting "OK" with the list only in the status bar + # leaves the IP box untouched. + self.connection_panel.set_ips([str(item) for item in result]) + elif message: self.bus_data_panel.set_result(message) text = f"{name} {t('action.ok')}".strip() self.statusBar().showMessage(text) diff --git a/apps/autocal/tests/test_controller_sequence.py b/apps/autocal/tests/test_controller_sequence.py new file mode 100644 index 0000000..296a134 --- /dev/null +++ b/apps/autocal/tests/test_controller_sequence.py @@ -0,0 +1,286 @@ +"""The controller-command sequence: what is actually sent, and when. + +These are the regressions that hardware found and the UI suite could not: + +* ``ZAux_Execute`` must be handed a real reply buffer. Passing ``NULL`` is not + an error the library reports — it writes through the null pointer and takes the + process down, so the assertion has to be on the *arguments*. +* Re-initialisation must stop and start the task the program actually runs on, + and a firmware refusal that arrives only in the reply text must not be + reported as a re-initialisation in progress. +* A session that adopts a program already running on the controller must still + be able to re-initialise it. + +No test here touches a controller or a DLL. +""" + +from __future__ import annotations + +import ctypes +import time +from pathlib import Path + +import pytest + +from orisys_autocal.motion import ( + DEFAULT_BAS_TASK, + INIT_OK, + INIT_RUNNING, + ConnectionSpec, + DemoTransport, + MotionService, + clamp_bas_task, +) +from orisys_autocal.motion.transport import _run_filename + +#: The shipped programs, used as the file the service is asked to download. +#: Real files rather than per-test temporaries: pytest's ``tmp_path`` needs to +#: enumerate a directory outside the workspace, which a confined run denies. +_BAS_DIR = Path(__file__).resolve().parents[1] / "config" / "bas" +SHIPPED_ECAT = _BAS_DIR / "ECAT.bas" +SHIPPED_RTEX = _BAS_DIR / "RTEX.bas" + +# ------------------------------------------------------------------ D1: the buffer + + +class _RecordingFunc: + """A ``ZAux_Execute`` double: records its arguments, writes a reply.""" + + def __init__(self, reply: bytes = b"", code: int = 0) -> None: + self.calls: list[tuple] = [] + self.reply = reply + self.code = code + + def __call__(self, handle, command, response, length): + size = int(getattr(length, "value", length)) + self.calls.append((handle, command, response, size)) + if response is None: + raise AssertionError("ZAux_Execute was called with a NULL reply buffer") + if self.reply: + ctypes.memmove(response, self.reply, min(len(self.reply), size - 1)) + return self.code + + +class _FakeDll: + def __init__(self, execute: _RecordingFunc) -> None: + self.ZAux_Execute = execute + + +@pytest.fixture +def client_with_fake_dll(): + from orisys_app_shared.zmotion import ZmotionClient + + client = ZmotionClient() + execute = _RecordingFunc(reply=b"Online command error, not find.\n") + client._dll = _FakeDll(execute) # noqa: SLF001 - the seam under test + client._handle = ctypes.c_void_p(0x1234) # noqa: SLF001 - pretend connected + return client, execute + + +def test_execute_passes_a_real_reply_buffer(client_with_fake_dll): + """The null-buffer bug: a NULL psResponse is a native crash, not an error.""" + client, execute = client_with_fake_dll + + result = client.execute('RUN "ECAT.bas",4') + + assert result.code == 0 + _handle, command, response, length = execute.calls[0] + assert command == b'RUN "ECAT.bas",4' + assert response is not None + assert length >= 1024 + + +def test_execute_returns_the_controllers_reply_text(client_with_fake_dll): + """The firmware explains a refused RUN only in the reply.""" + client, _execute = client_with_fake_dll + + result = client.execute('RUN "nope.bas",4') + + assert "not find" in str(result.value) + + +def test_transport_execute_hands_the_reply_back(): + from orisys_autocal.motion.transport import NativeTransport + + class _StubClient: + dll_dir = "stub" + + def execute(self, command): + from orisys_app_shared.zmotion import ZmotionValue + + return ZmotionValue(0, "controller says hello") + + assert NativeTransport(_StubClient()).execute("?X") == "controller says hello" + + +# ------------------------------------------------- D2: the re-init command shape + + +class RecordingTransport(DemoTransport): + """Demo platform that keeps every online command it was sent.""" + + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self.commands: list[str] = [] + + def execute(self, command: str) -> str: + self.commands.append(command) + return super().execute(command) + + +def _initialise(service, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline and service.bus.init_status != INIT_OK: + service.poll([0]) + time.sleep(0.02) + + +def test_reinit_stops_and_runs_the_configured_task(): + """Task 2 is not the vendor's task; stopping it restarts nothing.""" + transport = RecordingTransport() + service = MotionService(transport, bas_task=4) + service.connect(ConnectionSpec(mode="demo")) + service.download_bas(str(SHIPPED_ECAT), 1) + _initialise(service) + + transport.commands.clear() + result = service.reinit_bus() + + assert result.ok, result.message + assert transport.commands == ["STOPTASK 4", 'RUN "ECAT.bas",4'] + assert service.bus.init_status == INIT_RUNNING + + +def test_reinit_task_number_comes_from_configuration(): + transport = RecordingTransport() + service = MotionService(transport, bas_task=6) + service.connect(ConnectionSpec(mode="demo")) + service.download_bas(str(SHIPPED_RTEX), 1) + _initialise(service) + + transport.commands.clear() + assert service.reinit_bus().ok + assert transport.commands == ["STOPTASK 6", 'RUN "RTEX.bas",6'] + + +def test_a_controller_refusal_in_the_reply_is_not_reported_as_running(): + """``ZAux_Execute`` returns 0 for a RUN naming a program it does not have.""" + transport = RecordingTransport() + service = MotionService(transport, bas_task=4) + service.connect(ConnectionSpec(mode="demo")) + service.download_bas(str(SHIPPED_ECAT), 1) + _initialise(service) + + # The controller has ECAT.bas; pretend the operator's file was renamed. + transport._bas_file = "something-else.bas" # noqa: SLF001 - the stub's state + result = service.reinit_bus() + + assert not result.ok + assert "not find" in result.message + # And the bus must not be left claiming an initialisation is under way. + assert service.bus.init_status != INIT_RUNNING + + +# --------------------------------------------------- D3: the adopted-program path + + +def test_reinit_works_for_a_program_the_session_adopted(): + """The controller does not report the file name; the UI's remembered path does.""" + + class AdoptedProgram(RecordingTransport): + def user_var(self, name): + return {"BUS_TYPE": 0.0, "Bus_TotalAxisnum": 4.0, "Bus_InitStatus": 1.0}[name] + + transport = AdoptedProgram() + service = MotionService(transport, bas_task=4) + assert service.connect( + ConnectionSpec(mode="demo"), bas_path=r"C:\somewhere\ECAT.bas" + ).ok + + # This stub never changes its answer, so let the 25 s settle window lapse + # rather than waiting for it: what is under test is the command, not the wait. + service._init_timeout = 0.01 # noqa: SLF001 - shortening the vendor window + time.sleep(0.02) + + transport.commands.clear() + result = service.reinit_bus() + + assert result.ok, result.message + assert transport.commands == ["STOPTASK 4", 'RUN "ECAT.bas",4'] + + +def test_adopted_program_without_a_known_file_says_so(): + """The old message claimed no program was loaded while the bus panel said OK.""" + + class AdoptedProgram(DemoTransport): + def user_var(self, name): + return {"BUS_TYPE": 0.0, "Bus_TotalAxisnum": 4.0, "Bus_InitStatus": 1.0}[name] + + service = MotionService(AdoptedProgram()) + assert service.connect(ConnectionSpec(mode="demo")).ok + + result = service.reinit_bus() + + assert not result.ok + assert "文件名" in result.message or "file name" in result.message + + +def test_reinit_needs_the_bus_not_to_be_mid_run(): + service = MotionService(DemoTransport()) + service.connect(ConnectionSpec(mode="demo")) + service.download_bas(str(SHIPPED_ECAT), 1) + + assert not service.reinit_bus().ok # still inside the demo's init delay + + +# ------------------------------------------------------- task-number hygiene + + +@pytest.mark.parametrize("value", [0, -1, 9, 100, None, "x", ""]) +def test_a_nonsense_task_number_falls_back_to_the_vendor_default(value): + assert clamp_bas_task(value) == DEFAULT_BAS_TASK + + +@pytest.mark.parametrize("value", [1, 2, 4, 8, "4"]) +def test_a_plausible_task_number_survives(value): + assert clamp_bas_task(value) == int(value) + + +@pytest.mark.parametrize( + "command,expected", + [ + ('RUN "ECAT.bas",4', "ECAT.bas"), + ('RUN "a/b c/R TEX.bas",2', "a/b c/R TEX.bas"), + ("STOPTASK 4", ""), + ], +) +def test_run_filename_parsing(command, expected): + assert _run_filename(command) == expected + + +# ------------------------------------------------------- the simulated controller + + +def test_the_demo_rejects_a_program_it_does_not_have(): + """The simulation must misbehave the way the firmware does, or tests lie.""" + transport = DemoTransport() + transport.open(ConnectionSpec(mode="demo")) + transport.bas_download(str(SHIPPED_ECAT), 1) + + assert transport.execute('RUN "ECAT.bas",4') == "" + assert "not find" in transport.execute('RUN "other.bas",4') + + +def test_the_demo_keeps_the_downloaded_file_name(): + transport = DemoTransport() + transport.open(ConnectionSpec(mode="demo")) + transport.bas_download(str(SHIPPED_RTEX), 1) + + assert transport._bas_file == "RTEX.bas" # noqa: SLF001 - the stub's state + + +def test_a_missing_connection_is_still_reported(): + service = MotionService(DemoTransport()) + result = service.reinit_bus() + assert not result.ok + assert result.message diff --git a/apps/autocal/tests/test_errors.py b/apps/autocal/tests/test_errors.py index 8395311..0a3a005 100644 --- a/apps/autocal/tests/test_errors.py +++ b/apps/autocal/tests/test_errors.py @@ -52,3 +52,45 @@ def test_an_unknown_code_points_at_zdevelop(): def test_a_documented_code_gets_no_hint_appended(): assert errors.describe_with_hint(30000) == errors.describe(30000) + + +# ------------------------------------------------- host-library (PC) codes + + +def test_a_host_library_code_is_not_sent_to_the_firmware_table(): + """``20008`` is not a controller code: ZDevelop cannot have it. + + It is what ``ZAux_OpenEth`` returned on the first, transient attempt at the + bench, and the old text sent the operator to a table that cannot contain it. + """ + message = errors.describe_with_hint(20008) + + assert "20008" in message + assert errors.UNKNOWN_HINT not in message + assert "上位机" in message + assert errors.HOST_LIB_HINT_ZH in message + + +def test_the_host_library_hint_is_localised(): + zh = errors.describe_with_hint(20008) + en = errors.describe_with_hint(20008, language="en_US") + + assert errors.HOST_LIB_HINT_EN in en + assert zh != en + + +@pytest.mark.parametrize("code", [20000, 20008, 25000, 29999]) +def test_the_host_library_range_is_recognised(code): + assert errors.is_host_lib_error(code) + assert "错误码" not in errors.describe(code) + + +@pytest.mark.parametrize("code", [19999, 30000, 30005, 12345, 222, 0]) +def test_codes_outside_the_host_range_are_not_labelled_as_library_faults(code): + assert not errors.is_host_lib_error(code) + + +def test_a_controller_code_keeps_the_firmware_hint(): + """222 is firmware (``Other error``), so ZDevelop is still the right pointer.""" + message = errors.describe_with_hint(222) + assert errors.UNKNOWN_HINT in message diff --git a/apps/autocal/tests/test_locales_autocal.py b/apps/autocal/tests/test_locales_autocal.py index f82abf8..21c9449 100644 --- a/apps/autocal/tests/test_locales_autocal.py +++ b/apps/autocal/tests/test_locales_autocal.py @@ -79,6 +79,16 @@ def test_the_default_preferences_file_matches_the_schema_shape(): assert key in schema, f"preferences.default.json has an unknown key: {key}" +def test_the_shipped_defaults_carry_the_controller_task_number(): + """Re-initialisation names a task; if this key is missing every re-init is wrong.""" + from orisys_autocal.motion import DEFAULT_BAS_TASK + + shipped = json.loads( + (LOCALES.parent / "preferences.default.json").read_text(encoding="utf-8") + ) + assert shipped["bas_task"] == DEFAULT_BAS_TASK == 4 + + def _walk(node, prefix: str = ""): for key, value in node.items(): path = f"{prefix}{key}" diff --git a/apps/autocal/tests/test_units.py b/apps/autocal/tests/test_units.py index 4bf5161..1352903 100644 --- a/apps/autocal/tests/test_units.py +++ b/apps/autocal/tests/test_units.py @@ -1,11 +1,13 @@ -"""Unit-to-millimetre conversion. +"""Unit conversion: linear mm and rotary degrees. -The reference program carries the bench measurement in a comment:: +Bench notes:: # 轴0(X)/轴1(Y) 实测: UNITS=10 时 1000用户单位 = 2mm → 每毫米 5000 脉冲 # 当量5 → 1000单位=1mm,不是100单位=1mm + # 轴2: 12500 用户单位 = 90° + # 轴3: 29500 用户单位 = 90° -These are the assertions that keep that comment honest. +These are the assertions that keep those notes honest. """ from __future__ import annotations @@ -16,8 +18,12 @@ import pytest from orisys_autocal.motion.units import ( PULSES_PER_MM, + ROTARY_UNITS_PER_90_DEG, + deg_to_user_units, mm_to_user_units, safe_float, + units_per_degree, + user_units_to_deg, user_units_to_mm, ) @@ -71,3 +77,35 @@ def test_zero_falls_back_because_it_would_divide(text): @pytest.mark.parametrize(("text", "expected"), [("5", 5.0), (" 2.5 ", 2.5), ("-3", -3.0)]) def test_usable_entries_are_parsed(text, expected): assert safe_float(text, 42.0) == expected + + +def test_rotary_bench_constants(): + assert ROTARY_UNITS_PER_90_DEG[2] == 12500.0 + assert ROTARY_UNITS_PER_90_DEG[3] == 29500.0 + + +@pytest.mark.parametrize( + ("axis", "user_units", "expected_deg"), + [ + (2, 12500.0, 90.0), + (2, 25000.0, 180.0), + (3, 29500.0, 90.0), + (3, 14750.0, 45.0), + (2, 0.0, 0.0), + (3, -29500.0, -90.0), + ], +) +def test_rotary_conversion_matches_the_bench(axis, user_units, expected_deg): + assert user_units_to_deg(axis, user_units) == pytest.approx(expected_deg) + assert deg_to_user_units(axis, expected_deg) == pytest.approx(user_units) + + +def test_units_per_degree_is_units_over_ninety(): + assert units_per_degree(2) == pytest.approx(12500.0 / 90.0) + assert units_per_degree(3) == pytest.approx(29500.0 / 90.0) + + +def test_uncalibrated_rotary_axis_is_not_a_number(): + assert math.isnan(units_per_degree(0)) + assert math.isnan(user_units_to_deg(1, 1000.0)) + assert math.isnan(deg_to_user_units(4, 90.0)) diff --git a/apps/autocal/tools/home_trace.py b/apps/autocal/tools/home_trace.py new file mode 100644 index 0000000..ae598ea --- /dev/null +++ b/apps/autocal/tools/home_trace.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Watch one axis's drive frame while you start homing from elsewhere. + +Homing failures are hard to read after the fact: the drive reports a state, the +controller reports another, and by the time anyone looks, the motion has stopped +and the only evidence is "it ended up in the wrong place". This samples the +whole per-axis frame — commanded and measured position, the drive status word +(including the limit/home switch bits and the homing bits), the controller's +own motion and homing state, and the three mapped inputs — and prints a line +only when something changes. + +Run it, then press 启动回零 in the app (or send the homing command from ZDevelop) +while it is sampling: + + python apps\\autocal\\tools\\home_trace.py --axis 3 --seconds 45 + python apps\\autocal\\tools\\home_trace.py --axis 3 --seconds 45 --csv trace.csv + +Reading is non-mutating: it never enables, moves, homes or stops an axis. It +does open its own connection to the controller, so run it *instead of* having +the app open if two clients on one controller is a problem on your bench; the +ZMotion runtime generally allows it, but the app's own poller is the only writer +you want during a homing run. + +What the lines mean (CiA-402 status word, exposed here as ``DRIVE_STATUS``): + +====== ================================================== +bit 1 positive limit switch active +bit 2 negative limit switch active +bit 3 drive fault +bit 4 home (datum) switch active +bit 10 target reached +bit 12 homing attained — the drive *thinks* it found its reference +bit 13 homing error +====== ================================================== + +A homing run that ends with bit 12 set and no error, at a position you did not +expect, means the drive accepted a reference the mechanical setup does not +agree with — usually a switch polarity the drive reads as "already made", not a +missing index pulse (which instead creeps until something stops it). + +Exit status is 0 whether or not a run was seen: this is an observation tool, and +"nothing happened" is a result. +""" + +from __future__ import annotations + +import argparse +import csv +import ctypes +import os +import sys +import time +from pathlib import Path + +APP_ROOT = Path(__file__).resolve().parents[1] +for path in (APP_ROOT / "src", APP_ROOT.parents[1] / "shared" / "src"): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +os.environ.setdefault("ORISYS_APP_ID", "autocal") + +#: Fields sampled per tick, in the order the header and CSV are written. +FIELDS = ( + "t", + "dpos", + "mpos", + "idle", + "drive_status", + "p_ot", + "n_ot", + "fault", + "home_sw", + "target_reached", + "homing_attained", + "homing_error", + "axis_status", + "home_status", + "in_rev", + "in_fwd", + "in_datum", +) + +#: Position change above this counts as motion, in user units. +MOVE_EPSILON = 1e-6 + + +def _bit(value: int, index: int) -> int: + return (int(value) >> index) & 1 + + +class AxisReader: + """The few controller calls this tool needs, on one connection.""" + + def __init__(self, ip: str) -> None: + from orisys_app_shared.zmotion import ZmotionClient + + self._client = ZmotionClient().load() + result = self._client.open_eth(ip) + result.raise_for_status(f"open {ip}") + self._handle = self._client._require_handle() # noqa: SLF001 - one client + + def close(self) -> None: + self._client.close() + + def _param(self, name: str, axis: int) -> float | None: + value = ctypes.c_float() + code = self._client._dll.ZAux_Direct_GetParam( # noqa: SLF001 + self._handle, name.encode(), ctypes.c_int(axis), ctypes.byref(value) + ) + return float(value.value) if code == 0 else None + + def _input(self, pin: int) -> int | None: + value = ctypes.c_int() + code = self._client._dll.ZAux_Direct_GetIn( # noqa: SLF001 + self._handle, ctypes.c_int(pin), ctypes.byref(value) + ) + return int(value.value) if code == 0 else None + + def sample(self, axis: int, elapsed: float) -> dict: + raw_status = self._param("DRIVE_STATUS", axis) + status = int(raw_status or 0) & 0xFFFF + # The three switch inputs are mapped contiguously by ECAT.bas / RTEX.bas: + # REV at 128 + 16*axis, then FWD and DATUM. + base = 128 + 16 * axis + return { + "t": round(elapsed, 3), + "dpos": self._param("DPOS", axis), + "mpos": self._param("MPOS", axis), + "idle": self._param("IDLE", axis), + "drive_status": f"0x{status:04x}", + "p_ot": _bit(status, 1), + "n_ot": _bit(status, 2), + "fault": _bit(status, 3), + "home_sw": _bit(status, 4), + "target_reached": _bit(status, 10), + "homing_attained": _bit(status, 12), + "homing_error": _bit(status, 13), + "axis_status": self._client.get_axis_status(axis).value_or_none, + "home_status": self._client.bus_home_status(axis).value_or_none, + "in_rev": self._input(base), + "in_fwd": self._input(base + 1), + "in_datum": self._input(base + 2), + } + + +def _render(row: dict) -> str: + return ( + f"t={row['t']:7.3f} {row['drive_status']} " + f"OT+={row['p_ot']} OT-={row['n_ot']} HOME={row['home_sw']} FAULT={row['fault']} " + f"REACHED={row['target_reached']} HOMED={row['homing_attained']} " + f"HOMERR={row['homing_error']} " + f"DPOS={row['dpos']} IDLE={row['idle']} ctrl_home_status={row['home_status']} " + f"IN={row['in_rev']}/{row['in_fwd']}/{row['in_datum']}" + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="home-trace", description=__doc__) + parser.add_argument("--axis", type=int, required=True, help="axis number to watch") + parser.add_argument("--seconds", type=float, default=30.0, help="how long to sample") + parser.add_argument("--ip", default="192.168.0.11", help="controller address") + parser.add_argument("--interval-ms", type=float, default=20.0, help="sample period") + parser.add_argument("--csv", default="", help="also write every sample to this CSV") + args = parser.parse_args(argv) + + reader = AxisReader(args.ip) + print(f"watching axis {args.axis} for {args.seconds:g}s — start homing now", flush=True) + period = max(5.0, args.interval_ms) / 1000.0 + deadline = time.monotonic() + args.seconds + started = time.monotonic() + + rows: list[dict] = [] + previous: dict | None = None + movements = 0 + try: + while time.monotonic() < deadline: + row = reader.sample(args.axis, time.monotonic() - started) + rows.append(row) + interesting = ("drive_status", "idle", "p_ot", "n_ot", "home_sw", "fault", + "target_reached", "homing_attained", "homing_error", + "axis_status", "home_status", "in_rev", "in_fwd", "in_datum") + changed = previous is None or any(row[k] != previous[k] for k in interesting) + moved = previous is not None and row["mpos"] is not None and previous["mpos"] is not None \ + and abs(row["mpos"] - previous["mpos"]) > MOVE_EPSILON + if moved and not changed: + movements += 1 + if movements % 25 == 0: # a heartbeat while it travels + print(" " + _render(row), flush=True) + elif changed: + print(" " + _render(row), flush=True) + previous = row + time.sleep(period) + except KeyboardInterrupt: # pragma: no cover - operator convenience + print("\ninterrupted", flush=True) + finally: + reader.close() + + if args.csv: + path = Path(args.csv) + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=list(FIELDS)) + writer.writeheader() + writer.writerows(rows) + print(f"wrote {len(rows)} samples to {path}", flush=True) + + if previous is not None: + print("final: " + _render(previous), flush=True) + if previous["homing_attained"] and not previous["homing_error"]: + print( + "note: the drive reports homing attained — if the position is not where the " + "mechanics say home is, the switch polarity the drive sees is wrong (it read " + "'already made'), or the wrong switch edge was chosen.", + flush=True, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/apps/autocal/tools/switch_probe.py b/apps/autocal/tools/switch_probe.py new file mode 100644 index 0000000..7ce10e7 --- /dev/null +++ b/apps/autocal/tools/switch_probe.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Does the switch signal actually toggle? Sample the bits, count the edges. + +``home_trace.py`` prints a line whenever anything about the axis changes, which +is right for watching a homing run. This does the opposite job: it samples as +fast as the controller answers, counts every transition of the switch bits, and +reports them as a total. That turns "I think the switch works" into a number. + +Use it to answer one question — *does this switch reach the drive at all?* — for +a rotary axis where a partial move may simply not cross the switch: + + # drive the axis through at least one full revolution while this runs + python apps\\autocal\\tools\\switch_probe.py --axis 3 --seconds 60 + python apps\\autocal\\tools\\switch_probe.py --axis 3 --seconds 60 --csv edges3.csv + +Interpretation: + +* **``HOME`` edges > 0** — the bit follows the switch. Its polarity is the + ``rest→active`` pair printed in the summary; pick the homing method whose + edge matches. +* **``HOME`` edges = 0 while ``travel`` covers a full revolution** — the signal + never reached the drive. On a fail-safe input wired normally closed, a broken + wire looks exactly like "switch made" (the drive reads one constant level), so + zero edges after a full turn is a *wiring* finding, not a homing-method one. + Change methods all you like; no CiA-402 method can use a signal the drive + does not see. + +Reading is non-mutating: it never enables, moves, homes or stops anything. You +drive the axis (app jog, MPG, or hand), this only listens. + +Exit status is 0 in all outcomes — including "no edges", which is a result. +""" + +from __future__ import annotations + +import argparse +import csv +import ctypes +import os +import sys +import time +from pathlib import Path + +APP_ROOT = Path(__file__).resolve().parents[1] +for path in (APP_ROOT / "src", APP_ROOT.parents[1] / "shared" / "src"): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +os.environ.setdefault("ORISYS_APP_ID", "autocal") + +#: The bits worth counting, as ``label -> (bit index in DRIVE_STATUS, description)``. +WATCHED_BITS = { + "P_OT": (1, "positive limit switch"), + "N_OT": (2, "negative limit switch"), + "FAULT": (3, "drive fault"), + "HOME": (4, "home / datum switch"), +} + +#: Reported alongside the drive bits, because the controller inverts these and +#: the pair is what tells a wiring fault from a working switch. +INPUT_OFFSETS = {"IN_REV": 0, "IN_FWD": 1, "IN_DATUM": 2} # from 128 + 16*axis + + +class Sampler: + def __init__(self, ip: str) -> None: + from orisys_app_shared.zmotion import ZmotionClient + + self._client = ZmotionClient().load() + self._client.open_eth(ip).raise_for_status(f"open {ip}") + self._handle = self._client._require_handle() # noqa: SLF001 - one client + + def close(self) -> None: + self._client.close() + + def param(self, name: str, axis: int) -> float | None: + value = ctypes.c_float() + code = self._client._dll.ZAux_Direct_GetParam( # noqa: SLF001 + self._handle, name.encode(), ctypes.c_int(axis), ctypes.byref(value) + ) + return float(value.value) if code == 0 else None + + def input(self, pin: int) -> int | None: + value = ctypes.c_int() + code = self._client._dll.ZAux_Direct_GetIn( # noqa: SLF001 + self._handle, ctypes.c_int(pin), ctypes.byref(value) + ) + return int(value.value) if code == 0 else None + + def now(self, axis: int) -> tuple[int, float | None, dict[str, int | None]]: + status = int(self.param("DRIVE_STATUS", axis) or 0) & 0xFFFF + base = 128 + 16 * axis + inputs = {name: self.input(base + offset) for name, offset in INPUT_OFFSETS.items()} + return status, self.param("MPOS", axis), inputs + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="switch-probe", description=__doc__) + parser.add_argument("--axis", type=int, required=True) + parser.add_argument("--seconds", type=float, default=60.0) + parser.add_argument("--ip", default="192.168.0.11") + parser.add_argument("--csv", default="", help="write the edge log to this CSV") + args = parser.parse_args(argv) + + sampler = Sampler(args.ip) + print( + f"counting edges on axis {args.axis} for {args.seconds:g}s — " + f"drive the axis through a full revolution now", + flush=True, + ) + + deadline = time.monotonic() + args.seconds + edges: dict[str, list[tuple[float, int, int]]] = {name: [] for name in WATCHED_BITS} + input_edges: dict[str, list[tuple[float, int, int]]] = {name: [] for name in INPUT_OFFSETS} + counts: dict[str, int] = {name: 0 for name in WATCHED_BITS} + input_counts: dict[str, int] = {name: 0 for name in INPUT_OFFSETS} + previous: dict | None = None + travel = 0.0 + samples = 0 + events: list[dict] = [] + start = time.monotonic() + + def record(elapsed: float, status: int) -> None: + for label, (bit, _desc) in WATCHED_BITS.items(): + value = (status >> bit) & 1 + if previous is not None and value != ((previous["status"] >> bit) & 1): + counts[label] += 1 + edges[label].append((elapsed, (previous["status"] >> bit) & 1, value)) + events.append({"t": round(elapsed, 3), "kind": "drive_bit", "name": label, + "from": (previous["status"] >> bit) & 1, "to": value}) + + try: + while time.monotonic() < deadline: + status, mpos, inputs = sampler.now(args.axis) + elapsed = time.monotonic() - start + samples += 1 + if previous is not None and mpos is not None and previous["mpos"] is not None: + travel += abs(mpos - previous["mpos"]) + record(elapsed, status) + for label, value in inputs.items(): + if previous is not None and value != previous["inputs"][label]: + input_counts[label] += 1 + input_edges[label].append((elapsed, previous["inputs"][label], value)) + events.append({"t": round(elapsed, 3), "kind": "input", "name": label, + "from": previous["inputs"][label], "to": value}) + previous = {"status": status, "mpos": mpos, "inputs": inputs, "t": elapsed} + time.sleep(0.005) + except KeyboardInterrupt: # pragma: no cover - operator convenience + print("\ninterrupted", flush=True) + finally: + sampler.close() + + first_status = None + if samples: + first_status = previous["status"] + print() + print(f"samples: {samples} total travel: {travel:.1f} user units") + print("drive bit edges:") + for label, (bit, desc) in WATCHED_BITS.items(): + if counts[label]: + seq = " ".join(f"{a}->{b}" for _t, a, b in edges[label][:8]) + print(f" {label:6s} ({desc}): {counts[label]} edges first: {seq}") + else: + print(f" {label:6s} ({desc}): 0 edges — constant at {(previous['status'] >> bit) & 1 if samples else '?'}") + print("controller input edges (INVERT_IN applied):") + for label in INPUT_OFFSETS: + state = previous["inputs"][label] if samples else "?" + print(f" {label:8s}: {input_counts[label]} edges constant at {state}") + + if args.csv: + path = Path(args.csv) + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=["t", "kind", "name", "from", "to"]) + writer.writeheader() + writer.writerows(events) + print(f"wrote {len(events)} edge events to {path}") + + if samples and counts["HOME"] == 0 and counts["P_OT"] == 0 and counts["N_OT"] == 0: + print() + print("NOTE: no switch bit moved at all. If the travel above covers a full revolution,") + print(" the drive is not receiving a working switch signal — a broken wire on a") + print(" normally-closed fail-safe input reads exactly like a made switch. Check") + print(" the drive's DI wiring then, not the homing method.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/apps/studio/config/compound_eye.png b/apps/studio/config/compound_eye.png index 7ea4e12..3769363 100644 Binary files a/apps/studio/config/compound_eye.png and b/apps/studio/config/compound_eye.png differ diff --git a/shared/src/orisys_app_shared/zmotion/client.py b/shared/src/orisys_app_shared/zmotion/client.py index f62c42f..e07142b 100644 --- a/shared/src/orisys_app_shared/zmotion/client.py +++ b/shared/src/orisys_app_shared/zmotion/client.py @@ -21,6 +21,7 @@ from typing import Any # noqa: F401 (kept for ZmotionValue.value's annotation) from .loader import ZmotionDllNotFound, load_dll, resolve_dll_dir __all__ = [ + "RESPONSE_BUFFER_BYTES", "ZAuxError", "ZmotionClient", "ZmotionDllNotFound", @@ -28,6 +29,11 @@ __all__ = [ "ZmotionValue", ] +#: Reply buffer for :meth:`ZmotionClient.execute`. The vendor Python wrapper +#: allocates 2 048 bytes for the same call; the controller's diagnostics +#: (``?*ETHERCAT`` and friends) run to a few hundred. +RESPONSE_BUFFER_BYTES = 2048 + class ZAuxError(RuntimeError): """A ``ZAux_*`` call returned a non-zero code. @@ -215,14 +221,29 @@ class ZmotionClient: return ZmotionResult(int(ret)) def execute(self, command: str) -> ZmotionValue: - """Run an online command string (e.g. ``RUN "ECAT.bas",4``).""" + """Run an online command string (e.g. ``RUN "ECAT.bas",4``). + + Returns the controller's own reply text as the value. The reply buffer + is not optional: ``ZAux_Execute`` writes into ``psResponse`` for any + command that answers — a ``?`` query does — and passing ``NULL`` with a + zero length is a write through a null pointer inside the DLL, which + terminates the process rather than raising. The vendor wrapper allocates + the same 2 048-byte buffer for this reason. + + A non-zero code is *not* the only failure mode: an online command that + the firmware rejects (a ``RUN`` naming a program the controller does not + have) can still return 0 while putting the diagnosis in the reply text. + Callers that need to distinguish that must inspect the returned value. + """ + buffer = (ctypes.c_char * RESPONSE_BUFFER_BYTES)() ret = self._dll.ZAux_Execute( # type: ignore[union-attr] self._require_handle(), command.encode("utf-8"), - None, - ctypes.c_uint32(0), + buffer, + ctypes.c_uint32(RESPONSE_BUFFER_BYTES), ) - return ZmotionValue(int(ret), command) + text = buffer.value.decode("utf-8", errors="replace") if buffer.value else "" + return ZmotionValue(int(ret), text.strip()) def get_user_var(self, name: str) -> ZmotionValue: """Read a global variable from the running ``.bas`` program.