test(focas): bring the vendored focas-mock onto the real 31i-B wire shapes; integration tests follow

The mock had served the shapes the driver once agreed with it on, so it validated the very defects just fixed: it ANSWERED the non-existent 0x00D3 cnc_getfigure, wrote base/exponent 0/0 into position blocks, put the macro decimal at offset 4, and served spindle 'max rpm' as a raw constant. Ported from the sister ignitionoee mock (local divergence recorded in VENDORED.md; the LE timer and alarm cap are kept): - 0x00D3 refused with EW_FUNC, as the real control does. - 0x0026 blocks carry base 10 / position_figures exponent; signed arg2, arg2 = 0 or past the axis count refused EW_ATTRIB. - cnc_rdmacro is the 8-byte ODBM shape, 9-digit mantissa, unset = vacant. - servo/spindle meter records carry the envelope; spindle meters are 1/16384-of-full-scale counts (motor speed = speed / max_rpm). Integration tests: spindle test reads Load/MotorSpeedPct percentages; the position-scale test now proves the block's own scale wins; new Vacant_positions_and_macros_read_bad_over_the_wire. 12/12 pass against the mock run locally (python -m focas_mock.cli serve, OTOPCUA_FOCAS_SIM_ENDPOINT).
2026-10-07 14:12:06 -04:00
parent 26b512adba
commit 9b51dbca60
5 changed files with 279 additions and 55 deletions
@@ -43,3 +43,18 @@ Copy-Item "$src\Dockerfile" "$dest\"
 Update the snapshot date at the top of this file afterward. No other
 files belong here — the Docker build context is just the Python package
 and its metadata.
+
+## Local divergence (do not lose on refresh)
+
+`src/focas_mock/server.py` and `defaults.py` carry OtOpcUa-local edits that make the wire
+shapes match a real FANUC 31i-B (Makino Z-34184) instead of shapes the driver and the mock had
+agreed on between themselves. A refresh that overwrites `src/` must re-apply them:
+
+- the little-endian `cnc_rdtimer` minute/msec encoding (2026-06-25, live-validated);
+- the alarm count cap on `cnc_rdalmmsg2`;
+- (2026-10-07, ported from the sister ignitionoee FOCAS mock) `0x00D3` ("cnc_getfigure", not a
+  real FOCAS wire command) refused with `EW_FUNC`; `0x0026` position blocks carrying their own
+  `base/exponent` (`position_figures` is now that exponent) and a signed `arg2`;
+  `cnc_rdmacro` as the 8-byte ODBM shape (`dec_val` at `[6..7]`, 9-digit mantissa, unset
+  macro = vacant); servo/spindle meter records with the `base/exponent` envelope, spindle
+  meters in 1/16384-of-full-scale counts.
@@ -78,9 +78,12 @@ def make_default_state(profile: Mapping[str, Any]) -> dict[str, Any]:
        "acts": 3200,
        "acts2": [3200] + [0] * (max_spindles - 1),
        "axis_names": axis_names[:max_axis],
-        # Per-axis position decimal places served by cnc_getfigure (command 0x00d3).
-        # Default every configured axis to 0 ⇒ 10^0 = 1.0 scale, leaving existing
-        # position reads unchanged. mock_patch can override individual axes.
+        # Per-axis position scale, served INSIDE each cnc_rddynamic2 (0x0026) block as
+        # base 10 / this exponent — NOT by a separate command (0x00d3 "cnc_getfigure" was
+        # never real; the mock refuses it with EW_FUNC like the control does). Default 0 ⇒
+        # 10^0 = 1.0, leaving existing position reads unchanged. mock_patch can override an
+        # axis with an int exponent, or a dict {"base": .., "exponent": ..} — e.g.
+        # {"exponent": 65535} for the vacant marker.
        "position_figures": {name: 0 for name in axis_names[:max_axis]},
        "spindle_names": spindle_names,
        "parameters": {
@@ -7,6 +7,7 @@ from typing import Any, Awaitable, Callable

 from .constants import (
    ADMIN_METHODS,
+    EW_ATTRIB,
    EW_FUNC,
    EW_HANDLE,
    EW_NUMBER,
@@ -24,7 +25,52 @@ from .profiles import load_profile
 Handler = Callable[[dict[str, Any]], Awaitable[tuple[int, dict[str, Any] | None, str | None]]]


+class WireRefusal(Exception):
+    """One response block refuses with this FOCAS ``rc``.
+
+    Raised from a payload builder when the COMMAND exists but its ARGUMENTS name nothing the
+    control would answer (``cnc_rddynamic2`` with ``arg2 = 0``, for instance). Command-level
+    refusals live in ``FocasMockServer._WIRE_REFUSED_COMMANDS``. Ported from the sister
+    ignitionoee mock (Focas.Wire-020).
+    """
+
+    def __init__(self, rc: int) -> None:
+        super().__init__(f"refused with rc={rc}")
+        self.rc = rc
+
+
 class FocasMockServer:
+    # OtOpcUa-local changes to this vendored snapshot (2026-10-07), ported from the sister
+    # ignitionoee FOCAS mock so the wire shapes match a real FANUC 31i-B (Makino Z-34184) rather
+    # than shapes the driver and the mock once agreed on between themselves:
+    #   - 0x00D3 ("cnc_getfigure") is refused with EW_FUNC, as the real control does;
+    #   - 0x0026 position blocks carry their own base/exponent;
+    #   - cnc_rdmacro is the 8-byte ODBM shape with dec_val at [6..7];
+    #   - servo/spindle meter records carry the base/exponent envelope, spindle meters in
+    #     1/16384-of-full-scale counts.
+    # The little-endian cnc_rdtimer encoding and the alarm count cap stay as they were.
+
+    # cnc_rdmacro (ODBM) encoding constants — see _wire_rdmacro / _normalize_macro.
+    _MACRO_ODBM_LENGTH = 10
+    _MACRO_VACANT_DEC = -1
+    _MACRO_SIGNIFICANT_DIGITS = 9
+
+    # Commands this mock REFUSES, and with which rc — the real control's answer. 0x00D3 does not
+    # exist in the FOCAS wire protocol; serving it let a driver validate a command the real
+    # 31i-B refuses with rc=1.
+    _WIRE_REFUSED_COMMANDS = {0xD3: EW_FUNC}
+
+    # Default base served inside a 0x0026 position block (every block of the live 31i-B
+    # capture reports base 10).
+    _POSITION_DEFAULT_BASE = 10
+
+    # Servo/spindle meter records carry the same base/exponent envelope (live: base 10, exp 0).
+    _METER_BASE = 10
+
+    # SPINDLE meters (0x0040) are a fraction of full scale in 1/16384 units: 16384 = 100% of
+    # rated load (metric 0) or of maximum spindle speed (metric 1). SERVO meters are direct %.
+    _METER_FULL_SCALE = 16384
+
    def __init__(
        self,
        host: str,
@@ -208,10 +254,22 @@ class FocasMockServer:

    def _wire_response_block(self, request_block: bytes) -> bytes:
        command = int.from_bytes(request_block[6:8], "big") if len(request_block) >= 8 else 0
-        payload = self._wire_payload(command, request_block)
+        refused_rc = self._WIRE_REFUSED_COMMANDS.get(command)
+        if refused_rc is not None:
+            return self._make_wire_response_block(command, b"", request_block, rc=refused_rc)
+        try:
+            payload = self._wire_payload(command, request_block)
+        except WireRefusal as refusal:
+            return self._make_wire_response_block(command, b"", request_block, rc=refusal.rc)
        return self._make_wire_response_block(command, payload, request_block)

-    def _make_wire_response_block(self, command: int, payload: bytes, request_block: bytes | None = None) -> bytes:
+    def _make_wire_response_block(
+        self,
+        command: int,
+        payload: bytes,
+        request_block: bytes | None = None,
+        rc: int = EW_OK,
+    ) -> bytes:
        block_len = 0x10 + len(payload)
        block = bytearray(block_len)
        block[0:2] = block_len.to_bytes(2, "big")
@@ -219,6 +277,7 @@ class FocasMockServer:
            block[2:8] = request_block[2:8]
        else:
            block[6:8] = command.to_bytes(2, "big")
+        block[8:10] = int(rc).to_bytes(2, "big", signed=True)
        block[14:16] = len(payload).to_bytes(2, "big")
        block[16:] = payload
        return bytes(block)
@@ -256,8 +315,6 @@ class FocasMockServer:
            return self._wire_spindle_metric(request_block)
        if command == 0x56:
            return self._wire_servo_meter()
-        if command == 0xD3:
-            return self._wire_position_figures()
        if command == 0x57:
            mode = self.store.snapshot()["operation_mode"]
            return self._u16(int(mode.get("mode", 0) if isinstance(mode, dict) else mode))
@@ -322,23 +379,54 @@ class FocasMockServer:
        return self._u32(value)

    def _wire_dynamic_axis(self, request_block: bytes) -> bytes:
+        """Serves `cnc_rddynamic2` (0x0026) positions, honouring a SIGNED `arg2` the way the real
+        31i-B does (probed 2026-08-12 by the sister ignitionoee project, Focas.Wire-017/020):
+        `arg2 = 1..n` -> exactly one 8-byte block for that axis; `arg2 = -1` -> every slot in one
+        block; `arg2 = 0` or past the axis count -> refused (EW_ATTRIB)."""
        selector = self._block_u32(request_block, 8)
-        axis = self._block_u32(request_block, 12)
+        axis = self._block_i32(request_block, 12)
        field = {4: "absolute", 1: "machine", 6: "relative", 7: "distance"}.get(selector, "absolute")
        state = self.store.snapshot()
        names = list(state["axis_names"])
        dynamic_axes = state["dynamic"]["axes"]
+        if axis == 0 or axis > len(names):
+            raise WireRefusal(EW_ATTRIB)
+        if axis > 0:
+            name = names[axis - 1]
+            return self._wire_position_block(state, name, dynamic_axes.get(name, {}).get(field, 0))
        payload = bytearray()
        for name in names:
-            position = dynamic_axes.get(name, {})
-            payload += self._u32(int(position.get(field, 0)))
-            payload += self._u32(0)
-        if axis > 0 and axis <= len(names):
-            name = names[axis - 1]
-            position = dynamic_axes.get(name, {})
-            return self._u32(int(position.get(field, 0))) + self._u32(0)
+            payload += self._wire_position_block(state, name, dynamic_axes.get(name, {}).get(field, 0))
        return bytes(payload)

+    def _wire_position_block(self, state: dict[str, Any], name: str, raw: int) -> bytes:
+        """Encodes one 8-byte `cnc_rddynamic2` position block: value, then its OWN scale.
+
+            [0..3] i32 raw value
+            [4..5] u16 base      (2 or 10)
+            [6..7] u16 exponent  (0xFFFF = the slot holds no valid position)
+
+        `position_figures[name]` is the exponent (base 10); a dict value
+        `{"base": .., "exponent": ..}` overrides the base too, so a profile can serve a
+        deliberately undecodable block and prove the client Bads that one axis node.
+        """
+        base, exponent = self._position_scale(state, name)
+        return (
+            self._u32(int(raw))
+            + (base & 0xFFFF).to_bytes(2, "big")
+            + (exponent & 0xFFFF).to_bytes(2, "big")
+        )
+
+    def _position_scale(self, state: dict[str, Any], name: str) -> tuple[int, int]:
+        """Returns the (base, exponent) pair `position_figures` configures for one axis."""
+        configured = state.get("position_figures", {}).get(name, 0)
+        if isinstance(configured, dict):
+            return (
+                int(configured.get("base", self._POSITION_DEFAULT_BASE)),
+                int(configured.get("exponent", 0)),
+            )
+        return self._POSITION_DEFAULT_BASE, int(configured)
+
    def _wire_wrparam(self, request_block: bytes) -> None:
        datano = self._block_u32(request_block, 8)
        extra = request_block[28:]
@@ -346,11 +434,38 @@ class FocasMockServer:
        self.store.merge_patch({"parameters": {str(datano): {"type": "long", "value": value, "decimal": 0}}})

    def _wire_rdmacro(self, request_block: bytes) -> bytes:
+        """Encodes an ODBM `cnc_rdmacro` (0x0015) response payload, as a real 31i-B sends it
+        (golden capture `ref-macro-500-0x0015.bin`: `05 f5 e1 00 00 0a 00 08` = 1.0):
+
+            [0..3]  i32 mantissa (normalised to 9 significant digits)
+            [4..5]  i16 length field — a CONSTANT 10
+            [6..7]  i16 dec_val — decimal places; 0xFFFF (-1) = VACANT
+
+        A macro the profile does not mock is emitted as vacant, as a real control answers an
+        undefined variable.
+        """
        number = self._block_u32(request_block, 8)
-        macro = self.store.snapshot()["macros"].get(str(number), {})
-        value = int(macro.get("value", 0)) if isinstance(macro, dict) else 0
-        decimal = int(macro.get("decimal", 0)) if isinstance(macro, dict) else 0
-        return self._u32(value) + self._u16(decimal) + self._u16(0)
+        macro = self.store.snapshot()["macros"].get(str(number))
+        if isinstance(macro, dict):
+            mantissa, dec_val = self._normalize_macro(
+                int(macro.get("value", 0)), int(macro.get("decimal", 0))
+            )
+        else:
+            mantissa, dec_val = 0, self._MACRO_VACANT_DEC
+        return self._u32(mantissa) + self._u16(self._MACRO_ODBM_LENGTH) + self._u16(dec_val)
+
+    def _normalize_macro(self, value: int, decimal: int) -> tuple[int, int]:
+        """Turns a configured fixed-point `(value, decimal)` into the `(mantissa, dec_val)` a CNC
+        returns: the mantissa padded to 9 significant digits, `dec_val` raised to match
+        (live: `#502 = 270` -> `270000000/10^6`). The real value is unchanged."""
+        if decimal < 0:
+            return 0, self._MACRO_VACANT_DEC
+        if value == 0:
+            return 0, decimal
+        pad = self._MACRO_SIGNIFICANT_DIGITS - len(str(abs(value)))
+        if pad <= 0:
+            return value, decimal
+        return value * (10**pad), decimal + pad

    def _wire_wrmacro(self, request_block: bytes) -> None:
        number = self._block_u32(request_block, 8)
@@ -377,31 +492,34 @@ class FocasMockServer:
        payload = bytearray()
        for name in state["axis_names"]:
            entry = configured.get(name, {"name": name, "value": 0})
+            # 8-byte record (live 31i-B): value, then the base/exponent envelope. No axis name —
+            # the client pairs names positionally from the companion 0x0089 block.
            payload += self._u32(int(entry.get("value", 0)))
-            payload += self._u16(0)
-            payload += self._u16(1)
-            payload += self._name_record(name, 4)
-        return bytes(payload)
-
-    def _wire_position_figures(self) -> bytes:
-        state = self.store.snapshot()
-        figures = state.get("position_figures", {})
-        payload = bytearray()
-        for name in state["axis_names"]:
-            payload += self._u16(int(figures.get(name, 0)))
+            payload += self._u16(self._METER_BASE)
+            payload += self._u16(int(entry.get("exponent", 0)))
        return bytes(payload)

    def _wire_spindle_metric(self, request_block: bytes) -> bytes:
+        # Both metrics are 1/16384-of-full-scale COUNTS on the wire, not the human units the
+        # config holds. Metric 0: load % of rated (16384 = 100%). Metric 1: motor speed as a
+        # fraction of the spindle's maximum (`speed / max_rpm`) — NOT the maximum itself.
        metric = self._block_u32(request_block, 8)
        spindle = self.store.snapshot()["spindle"]
-        if metric == 1:
-            values = [int(value) for value in spindle.get("max_rpm", [])]
-        else:
-            values = [int(entry.get("load", 0)) for entry in spindle.get("load", [])]
+        maxima = [int(value) for value in spindle.get("max_rpm", [])]
+        entries = spindle.get("load", [])
+        values = []
+        for index, entry in enumerate(entries):
+            if metric == 1:
+                maximum = maxima[index] if index < len(maxima) else 0
+                speed = int(entry.get("speed", 0))
+                values.append(round(speed * self._METER_FULL_SCALE / maximum) if maximum else 0)
+            else:
+                values.append(round(int(entry.get("load", 0)) * self._METER_FULL_SCALE / 100))
        payload = bytearray()
        for value in values[:4]:
            payload += self._u32(value)
-            payload += self._u32(0)
+            payload += self._u16(self._METER_BASE)
+            payload += self._u16(0)
        return bytes(payload)

    def _wire_program_name(self) -> bytes:
@@ -480,6 +598,12 @@ class FocasMockServer:
            return 0
        return int.from_bytes(block[offset : offset + 4], "big")

+    def _block_i32(self, block: bytes, offset: int) -> int:
+        """A SIGNED request argument — `-1` is a real selector (all axes / all spindles)."""
+        if offset + 4 > len(block):
+            return 0
+        return int.from_bytes(block[offset : offset + 4], "big", signed=True)
+
    def _u16(self, value: int) -> bytes:
        return int(value).to_bytes(2, "big", signed=True)