From 7e48e6f71f561620c6db8dc8b0106309f40ac1ad Mon Sep 17 00:00:00 2001 From: jepson2k <55201008+Jepson2k@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:19:26 -0400 Subject: [PATCH 1/4] Bound digital I/O requests and support named signal skills --- README.md | 10 ++++++ parol6/client/async_client.py | 41 +++++++++++++++++++---- parol6/client/dry_run_client.py | 20 ++++++++++- parol6/client/sync_client.py | 8 ++--- pyproject.toml | 2 +- tests/integration/test_digital_io.py | 50 ++++++++++++++++++++++++++++ 6 files changed, 118 insertions(+), 13 deletions(-) create mode 100644 tests/integration/test_digital_io.py diff --git a/README.md b/README.md index 673fbdd..c28f699 100644 --- a/README.md +++ b/README.md @@ -451,3 +451,13 @@ The existing `set_tcp_offset(x, y, z)` clears user rotation and now returns its queued index for confirmation. `tcp_offset()` still reads three translations; `tcp_transform()` reads all six values. Both raise `TimeoutError` when no valid reply arrives instead of reporting a misleading zero correction. + +Digital I/O reads and writes accept an optional per-call `timeout` in seconds: +`rbt.io(timeout=1.0)` returns `None` without a reply, while +`rbt.write_io(0, 1, timeout=1.0)` raises `TimeoutError` if acceptance remains +unconfirmed. The deadline includes transport setup and retries. Omitting it +retains the configured client timeout. The same options work on the sync client. +The client advertises `io.digital` for typed named-signal skills, which can be +imported from `waldo_commander.skills`; mappings are `waldoctl.signals.DigitalSignal` +values stored in a setup snapshot. Dry-run clients advertise `execution.preview` +so those skills require explicit observation fixtures during preview. diff --git a/parol6/client/async_client.py b/parol6/client/async_client.py index 4d6f723..b159124 100644 --- a/parol6/client/async_client.py +++ b/parol6/client/async_client.py @@ -5,6 +5,7 @@ import asyncio import contextlib import logging +import math import random import socket import struct @@ -257,7 +258,11 @@ class AsyncRobotClient(_RobotClientABC): @property def skill_capabilities(self) -> frozenset[str]: - return super().skill_capabilities | {"backend.parol6", "tool.gripper"} + return super().skill_capabilities | { + "backend.parol6", + "tool.gripper", + "io.digital", + } def __init__( self, @@ -850,16 +855,26 @@ async def angles(self) -> list[float] | None: resp = await self._request(AnglesCmd()) return resp.angles if isinstance(resp, AnglesResultStruct) else None - async def io(self) -> list[int] | None: + async def io(self, *, timeout: float | None = None) -> list[int] | None: """Digital I/O status [in1, in2, out1, out2, estop]. + ``timeout`` bounds setup, retries, and the reply; None uses client defaults. + Category: Query Example: io = rbt.io() """ - resp = await self._request(IOCmd()) - return resp.io if isinstance(resp, IOResultStruct) else None + if timeout is not None and ( + isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0 + ): + raise ValueError("I/O timeout must be positive and finite") + try: + async with asyncio.timeout(timeout): + resp = await self._request(IOCmd()) + return resp.io if isinstance(resp, IOResultStruct) else None + except TimeoutError: + return None async def joint_speeds(self) -> list[float] | None: """Current joint speeds in steps/sec [J1, J2, J3, J4, J5, J6]. @@ -1847,7 +1862,9 @@ async def jog_l( # --------------- IO / Gripper / Utility --------------- - async def write_io(self, index: int, value: int) -> int: + async def write_io( + self, index: int, value: int, *, timeout: float | None = None + ) -> int: """Set digital output by logical index (0 = first output pin). The firmware I/O byte layout is ``[in0, in1, out0, out1, estop, ...]`` @@ -1855,6 +1872,9 @@ async def write_io(self, index: int, value: int) -> int: Returns the command index (≥ 0) on success, -1 on failure. + ``timeout`` bounds command acceptance. TimeoutError leaves application + unconfirmed; None uses the client defaults. + Category: I/O Example: @@ -1866,8 +1886,15 @@ async def write_io(self, index: int, value: int) -> int: raise ValueError("I/O value must be 0 or 1") # Firmware bit layout: [in0, in1, out0, out1, estop, ...] firmware_index = index + 2 - result = await self._send(WriteIOCmd(port_index=firmware_index, value=value)) - return result + if timeout is not None and ( + isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0 + ): + raise ValueError("I/O timeout must be positive and finite") + async with asyncio.timeout(timeout): + result = await self._send( + WriteIOCmd(port_index=firmware_index, value=value) + ) + return result async def delay(self, seconds: float) -> int: """Insert a non-blocking delay in the motion queue. diff --git a/parol6/client/dry_run_client.py b/parol6/client/dry_run_client.py index 7890425..9387727 100644 --- a/parol6/client/dry_run_client.py +++ b/parol6/client/dry_run_client.py @@ -43,6 +43,7 @@ SelectToolCmd, SetTcpOffsetCmd, SetTcpTransformCmd, + WriteIOCmd, TeleportCmd, ToolActionCmd, ) @@ -546,7 +547,14 @@ def _simulate_cartesian_jog(self, cmd: JogLCommand) -> DryRunResult | None: @property def skill_capabilities(self) -> frozenset[str]: return frozenset( - {"motion.joint", "motion.linear", "tool.gripper", "backend.parol6"} + { + "motion.joint", + "motion.linear", + "tool.gripper", + "backend.parol6", + "io.digital", + "execution.preview", + } ) def angles(self) -> list[float]: @@ -601,6 +609,16 @@ def servo_j( return self._dispatch(build_cmd("servo_j_pose", pose, **kwargs)) return self._dispatch(build_cmd("servo_j", angles or [], **kwargs)) + def write_io(self, index: int, value: int, *, timeout: float | None = None) -> int: + if type(index) is not int or index not in (0, 1): + raise ValueError("Output index must be 0 or 1") + if type(value) not in (int, bool) or value not in (0, 1): + raise ValueError("Digital output must be 0 or 1") + result = self._dispatch(WriteIOCmd(port_index=index + 2, value=int(value))) + if result is not None and result.error is not None: + raise RuntimeError(str(result.error)) + return 0 + def delay(self, seconds: float = 0.0) -> None: pass diff --git a/parol6/client/sync_client.py b/parol6/client/sync_client.py index 1cc1054..b1d3598 100644 --- a/parol6/client/sync_client.py +++ b/parol6/client/sync_client.py @@ -287,13 +287,13 @@ def angles(self) -> list[float] | None: """ return _run(self._inner.angles()) - def io(self) -> list[int] | None: + def io(self, *, timeout: float | None = None) -> list[int] | None: """Digital I/O status. Returns: List of 5 integers [in1, in2, out1, out2, estop], or None on timeout. """ - return _run(self._inner.io()) + return _run(self._inner.io(timeout=timeout)) def joint_speeds(self) -> list[float] | None: """Current joint speeds in steps per second. @@ -851,9 +851,9 @@ def checkpoint(self, label: str) -> int: def wait_checkpoint(self, label: str, timeout: float = 30.0) -> bool: return _run(self._inner.wait_checkpoint(label, timeout=timeout)) - def write_io(self, index: int, value: int) -> int: + def write_io(self, index: int, value: int, *, timeout: float | None = None) -> int: """Set digital output by logical index (0 = first output pin).""" - return _run(self._inner.write_io(index, value)) + return _run(self._inner.write_io(index, value, timeout=timeout)) def delay(self, seconds: float) -> int: """Insert a non-blocking delay in the motion queue.""" diff --git a/pyproject.toml b/pyproject.toml index 0bc318d..c5d7648 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,7 +46,7 @@ dependencies = [ "psutil>=5.9", "msgspec>=0.18", "ormsgpack>=1.4.0", - "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.15.0", + "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.16.0", ] [tool.setuptools.packages.find] diff --git a/tests/integration/test_digital_io.py b/tests/integration/test_digital_io.py new file mode 100644 index 0000000..1a1fb5d --- /dev/null +++ b/tests/integration/test_digital_io.py @@ -0,0 +1,50 @@ +"""Digital I/O uses logical output indices and per-call reply deadlines.""" + +import asyncio +import socket +import time + +import pytest +from waldoctl.skills import skill + +from parol6 import AsyncRobotClient + + +def test_digital_io_readback_and_missing_peer_deadlines(client, server_proc): + before = client.io(timeout=2) + assert before is not None + try: + assert client.write_io(0, 1 - before[2], timeout=2) >= 0 + assert client.wait_status(lambda s: s.io[2] == 1 - before[2], timeout=2) + assert client.io(timeout=2)[2] == 1 - before[2] + finally: + client.write_io(0, before[2], timeout=2) + + async def missing_peer(): + @skill(id="test.io_deadline", version="1.0.0") + async def query(rbt): + return await rbt.io(timeout=0.05) + + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as silent: + silent.bind(("127.0.0.1", 0)) + async with AsyncRobotClient( + port=silent.getsockname()[1], timeout=5, retries=3 + ) as absent: + start = time.monotonic() + assert await query.async_call(absent) is None + assert time.monotonic() - start < 1.0, ( + "query ignored its per-call deadline" + ) + start = time.monotonic() + with pytest.raises(TimeoutError): + await absent.write_io(0, 1, timeout=0.05) + assert time.monotonic() - start < 1.0, ( + "write ignored its per-call deadline" + ) + for invalid in (0, -1, float("nan"), float("inf"), True): + with pytest.raises(ValueError): + await absent.io(timeout=invalid) + with pytest.raises(ValueError): + await absent.write_io(0, 1, timeout=invalid) + + asyncio.run(missing_peer()) From 8d3a266f816b1913735de44bc19e6b18350d7467 Mon Sep 17 00:00:00 2001 From: jepson2k <55201008+Jepson2k@users.noreply.github.com> Date: Fri, 11 Sep 2026 05:05:59 +0000 Subject: [PATCH 2/4] Drop stale replies before each request so a lapsed deadline cannot skew later queries A per-call deadline cancels a query after its datagram was sent; the reply then sits in the receive queue and, with no request ids on the wire, is handed to the next query of any type (returning None for its typed result and leaving every later query one reply behind) or, for a lapsed write_io, lets a stale index-less OK stand in for the next command's ack. Queued replies are discarded when a new request goes out, and a per-call deadline now bounds a single attempt instead of cancelling mid-receive. Co-Authored-By: Claude Fable 5.1 --- parol6/client/async_client.py | 49 ++++++++++++++++++++++------ tests/integration/test_digital_io.py | 29 ++++++++++++++++ 2 files changed, 68 insertions(+), 10 deletions(-) diff --git a/parol6/client/async_client.py b/parol6/client/async_client.py index b159124..4f71f88 100644 --- a/parol6/client/async_client.py +++ b/parol6/client/async_client.py @@ -588,7 +588,34 @@ async def _send(self, cmd: msgspec.Struct) -> int: self._transport.sendto(self._tx_buf) return 1 - async def _request(self, cmd: msgspec.Struct) -> Response | None: + def _drop_stale_replies(self) -> None: + """Discard replies already queued when a new request is about to go out. + + Nothing awaits them: their caller's deadline expired mid-flight. The + wire carries no request id, so handing one to the next request would + answer it with the wrong struct and leave every later query one reply + behind. + """ + kept = [] + while True: + try: + item = self._rx_queue.get_nowait() + except asyncio.QueueEmpty: + break + try: + stale = isinstance( + decode_message(item[0]), (ResponseMsg, OkMsg, ErrorMsg) + ) + except msgspec.DecodeError: + stale = False + if not stale: + kept.append(item) + for item in kept: + self._rx_queue.put_nowait(item) + + async def _request( + self, cmd: msgspec.Struct, timeout: float | None = None + ) -> Response | None: """Send a query command and wait for a typed response. Drains the receive queue until a ResponseMsg is found or timeout. @@ -596,6 +623,8 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None: Args: cmd: Typed command struct + timeout: Per-call deadline; when given, the query is sent once + with no retries so the deadline is the caller's total wait. Returns: Typed Response struct, or None on timeout. @@ -606,11 +635,14 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None: await self._ensure_endpoint() assert self._transport is not None data = encode_command(cmd) - for attempt in range(self.retries + 1): + wait = self.timeout if timeout is None else timeout + attempts = self.retries + 1 if timeout is None else 1 + for attempt in range(attempts): try: async with self._req_lock: + self._drop_stale_replies() self._transport.sendto(data) - end_time = time.monotonic() + self.timeout + end_time = time.monotonic() + wait while time.monotonic() < end_time: try: resp_data, _ = await asyncio.wait_for( @@ -637,7 +669,7 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None: pass except Exception: break - if attempt < self.retries: + if attempt < attempts - 1: backoff = min(0.5, 0.05 * (2**attempt)) + random.uniform(0, 0.05) await asyncio.sleep(backoff) return None @@ -657,6 +689,7 @@ async def _request_ok_raw(self, data: bytes, timeout: float) -> OkMsg: end_time = time.monotonic() + timeout async with self._req_lock: + self._drop_stale_replies() self._transport.sendto(data) while time.monotonic() < end_time: try: @@ -869,12 +902,8 @@ async def io(self, *, timeout: float | None = None) -> list[int] | None: isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0 ): raise ValueError("I/O timeout must be positive and finite") - try: - async with asyncio.timeout(timeout): - resp = await self._request(IOCmd()) - return resp.io if isinstance(resp, IOResultStruct) else None - except TimeoutError: - return None + resp = await self._request(IOCmd(), timeout=timeout) + return resp.io if isinstance(resp, IOResultStruct) else None async def joint_speeds(self) -> list[float] | None: """Current joint speeds in steps/sec [J1, J2, J3, J4, J5, J6]. diff --git a/tests/integration/test_digital_io.py b/tests/integration/test_digital_io.py index 1a1fb5d..f5117aa 100644 --- a/tests/integration/test_digital_io.py +++ b/tests/integration/test_digital_io.py @@ -48,3 +48,32 @@ async def query(rbt): await absent.write_io(0, 1, timeout=invalid) asyncio.run(missing_peer()) + + +def test_late_replies_never_answer_the_next_request(ports, server_proc): + """A reply that lands after its caller's deadline expired is not served + to the next request: the wire carries no request ids, so a stale reply + would answer the wrong query and leave every later one a reply behind.""" + from parol6.protocol.wire import IOResultStruct, pack_ok, pack_response + + async def scenario(): + async with AsyncRobotClient( + host=ports.server_ip, port=ports.server_port, timeout=5.0 + ) as rbt: + await rbt._ensure_endpoint() + peer = (ports.server_ip, ports.server_port) + rbt._rx_queue.put_nowait( + (pack_response(IOResultStruct(io=[0, 0, 0, 0, 1])), peer) + ) + pose = await rbt.pose() + assert pose is not None and len(pose) == 6 + assert await rbt.angles() is not None + rbt._rx_queue.put_nowait((pack_ok(), peer)) + index = await rbt.delay(0.1) + assert index >= 1, "a stale index-less OK must not stand in for the ack" + assert await rbt.wait_command(index, timeout=5) + for _ in range(5): + await rbt.io(timeout=1e-4) + assert await rbt.pose() is not None + + asyncio.run(scenario()) From 1901579f1de2a7d51d040bcbb06838e105bbd90d Mon Sep 17 00:00:00 2001 From: jepson2k <55201008+Jepson2k@users.noreply.github.com> Date: Fri, 11 Sep 2026 06:42:23 +0000 Subject: [PATCH 3/4] Keep the per-call I/O deadline over endpoint setup as well as the reply The per-call deadline moved inside _request, which no longer bounded endpoint setup and its retries: an io() against an absent peer waited the client's full retry budget. The outer deadline is back around the call; the inner one still keeps the query to a single attempt. Co-Authored-By: Claude Fable 5.1 --- parol6/client/async_client.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/parol6/client/async_client.py b/parol6/client/async_client.py index 4f71f88..f12f473 100644 --- a/parol6/client/async_client.py +++ b/parol6/client/async_client.py @@ -902,7 +902,13 @@ async def io(self, *, timeout: float | None = None) -> list[int] | None: isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0 ): raise ValueError("I/O timeout must be positive and finite") - resp = await self._request(IOCmd(), timeout=timeout) + # The outer deadline also bounds endpoint setup and its retries on an + # absent peer; the inner one keeps the query to a single attempt. + try: + async with asyncio.timeout(timeout): + resp = await self._request(IOCmd(), timeout=timeout) + except TimeoutError: + return None return resp.io if isinstance(resp, IOResultStruct) else None async def joint_speeds(self) -> list[float] | None: From a2e8194549ddf44b64b7bcffc7f8061a1227e4c2 Mon Sep 17 00:00:00 2001 From: jepson2k <55201008+Jepson2k@users.noreply.github.com> Date: Fri, 11 Sep 2026 06:44:09 +0000 Subject: [PATCH 4/4] Let the lapsed-deadline reply land before the next query in the stale-reply test Co-Authored-By: Claude Fable 5.1 --- tests/integration/test_digital_io.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/integration/test_digital_io.py b/tests/integration/test_digital_io.py index f5117aa..7e48c9c 100644 --- a/tests/integration/test_digital_io.py +++ b/tests/integration/test_digital_io.py @@ -72,8 +72,11 @@ async def scenario(): index = await rbt.delay(0.1) assert index >= 1, "a stale index-less OK must not stand in for the ack" assert await rbt.wait_command(index, timeout=5) - for _ in range(5): - await rbt.io(timeout=1e-4) - assert await rbt.pose() is not None + # A deadline that lapses mid-flight: the reply lands afterwards + # and must be dropped before the next query goes out. + assert await rbt.io(timeout=1e-4) is None + await asyncio.sleep(0.1) + pose = await rbt.pose() + assert pose is not None and len(pose) == 6 asyncio.run(scenario())