feat(modbus): Shared/RTU client 暴露 per-call timeout 參數#144
Open
csp0924 wants to merge 2 commits into
Open
Conversation
… timeout 參數 `SharedPymodbusTcpClient` 與 `PymodbusRtuClient` 各 8 個 read/write method 過去把 `timeout` 寫死在 `RequestQueueConfig.default_timeout`，雖然底層 `ModbusRequestQueue.submit()` 早已支援 `timeout: float | None = None`， wrapper 從未把這個參數 plumb 上去。結果 operator 只能在 queue 層級二選一： - default 太大：慢設備藏在 queue 內、CB 幾乎不會 trip - default 太小：對 slow-but-alive 設備 CB 誤觸發（H3 sandbox cliff 91.7% blackout） 本 PR 在所有 read/write method 加 optional `timeout: float | None = None` keyword，並 plumb 進 `_submit_request → queue.submit(timeout=...)`。 - 未指定 → 沿用 `config.default_timeout`（向後相容） - 指定 → 該 call 直接套用該 timeout，獨立於 queue 全域 default 容許 operator 對單一慢 unit 開高 timeout（如 PCS 2.5s），其他 fast unit 維持原來的 default 0.8s，CB 真正只在「server 不會回」場景才 trip。 Sandbox 驗證 `sandbox/modbus_per_call_timeout_demo.py`： - 全 default 0.8s（latency cliff 1.5s）→ pct_fail=100%（60/60 CB blackout） - per-call 覆寫 2.5s → pct_fail=0%，p99=1864ms 正常完成 純 additive，無 public API 變更語意（既有 caller 不傳 timeout 行為不變）。
There was a problem hiding this comment.
Pull request overview
This PR exposes a per-call timeout: float | None = None parameter on Modbus client read/write APIs and plumbs it down to the request-queue layer so operators can override RequestQueueConfig.default_timeout for specific slow calls while keeping the global default small.
Changes:
- Added
timeout: float | None = NonetoAsyncModbusClientBaseread/write abstract methods. - Plumbed
timeoutthroughSharedPymodbusTcpClient/PymodbusRtuClientwrappers intoModbusRequestQueue.submit(timeout=...). - Added a new test suite verifying per-call timeouts time out quickly and that omitting
timeoutpreserves the default behavior.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| tests/modbus/test_shared_client_timeout.py | Adds async tests to verify per-call timeout plumbing for Shared TCP and RTU clients and backward compatibility when timeout is omitted. |
| csp_lib/modbus/clients/client.py | Adds timeout support across client methods and passes it to queue submission for shared/RTU clients (also introduces per-call timeout behavior for PymodbusTcpClient). |
| csp_lib/modbus/clients/base.py | Updates the async Modbus client abstract interface to include the new timeout parameter. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| @abstractmethod | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1, timeout: float | None = None) -> list[bool]: |
Comment on lines
44
to
+62
| @abstractmethod | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1, timeout: float | None = None) -> list[bool]: | ||
| """ | ||
| 讀取線圈狀態 (Function Code 0x01) | ||
|
|
||
| Args: | ||
| address: 起始位址 | ||
| count: 讀取數量 | ||
| unit_id: 設備位址 (Slave ID) | ||
| timeout: 此請求逾時 (秒)；None 沿用 client / queue 預設值 | ||
|
|
||
| Returns: | ||
| 布林值列表 | ||
| """ | ||
|
|
||
| @abstractmethod | ||
| async def read_discrete_inputs(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| async def read_discrete_inputs( | ||
| self, address: int, count: int, unit_id: int = 1, timeout: float | None = None | ||
| ) -> list[bool]: |
Comment on lines
64
to
+147
| class PymodbusTcpClient(AsyncModbusClientBase): | ||
| """ | ||
| pymodbus TCP 非同步客戶端 | ||
|
|
||
| 使用 pymodbus 庫實作 Modbus TCP 通訊。 | ||
| 所有操作皆為非同步 (async/await)。 | ||
|
|
||
| 支援 pymodbus >= 3.0.0，自動適配 3.10.0+ API 變更。 | ||
|
|
||
| Args: | ||
| config: TCP 連線設定 | ||
|
|
||
| 使用範例： | ||
| >>> config = ModbusTcpConfig(host="192.168.1.100") | ||
| >>> async with PymodbusTcpClient(config) as client: | ||
| ... registers = await client.read_holding_registers(0, 10) | ||
| """ | ||
|
|
||
| def __init__(self, config: ModbusTcpConfig) -> None: | ||
| self._config = config | ||
| self._client: AsyncModbusTcpClient | None = None | ||
| # BUG-006：lock 防止併發 connect() 重複呼叫底層；以 client.connected 為準， | ||
| # 不用 sticky flag（避免網路掉線後 client.connected=False 卻被誤判已連線） | ||
| self._connect_lock = asyncio.Lock() | ||
|
|
||
| def _get_client(self) -> AsyncModbusTcpClient: | ||
| """取得或建立 pymodbus 客戶端""" | ||
| if self._client is None: | ||
| _ensure_pymodbus_imported() | ||
| assert _AsyncModbusTcpClient is not None # for type checker | ||
|
|
||
| self._client = _AsyncModbusTcpClient( | ||
| host=self._config.host, | ||
| port=self._config.port, | ||
| timeout=self._config.timeout, | ||
| retries=0, # 不重試 | ||
| ) | ||
| return self._client | ||
|
|
||
| async def connect(self) -> None: | ||
| """建立 TCP 連線（idempotent + 併發安全）""" | ||
| # BUG-006：lock 序列化併發呼叫；lock 內以 client.connected 為準判斷是否需 connect | ||
| # 確保網路掉線（client.connected 自動翻為 False）後仍可重連 | ||
| async with self._connect_lock: | ||
| client = self._get_client() | ||
| if client.connected: | ||
| return # 真實連線狀態已連，跳過底層呼叫 | ||
|
|
||
| logger.info(f"Connecting to {self._config.host}:{self._config.port}...") | ||
| connected = await client.connect() | ||
| if not connected: | ||
| logger.warning(f"Connection failed: {self._config.host}:{self._config.port}") | ||
| raise ModbusError(f"無法連線到 {self._config.host}:{self._config.port}") | ||
| logger.info(f"Connected to {self._config.host}:{self._config.port}") | ||
|
|
||
| async def disconnect(self) -> None: | ||
| """斷開 TCP 連線""" | ||
| if self._client is not None: | ||
| self._client.close() | ||
| logger.info(f"Disconnected from {self._config.host}:{self._config.port}") | ||
|
|
||
| async def is_connected(self) -> bool: | ||
| """檢查連線狀態""" | ||
| return self._client is not None and self._client.connected | ||
|
|
||
| @staticmethod | ||
| async def _maybe_wait_for(coro: Any, timeout: float | None) -> Any: | ||
| """若有指定 timeout, 用 asyncio.wait_for 包裝；否則直接 await coro。""" | ||
| if timeout is None: | ||
| return await coro | ||
| return await asyncio.wait_for(coro, timeout=timeout) | ||
|
|
||
| # ========== 讀取操作 ========== | ||
|
|
||
| async def read_coils(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| """讀取線圈狀態 (FC 0x01)""" | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1, timeout: float | None = None) -> list[bool]: | ||
| """讀取線圈狀態 (FC 0x01) | ||
|
|
||
| Args: | ||
| timeout: 此請求逾時 (秒)；None 沿用 config.timeout | ||
| """ | ||
| client = self._get_client() | ||
| response = await client.read_coils( | ||
| address=address, | ||
| count=count, | ||
| **_slave_kwarg(unit_id), | ||
| response = await self._maybe_wait_for( | ||
| client.read_coils(address=address, count=count, **_slave_kwarg(unit_id)), | ||
| timeout, |
Comment on lines
+129
to
199
| @staticmethod | ||
| async def _maybe_wait_for(coro: Any, timeout: float | None) -> Any: | ||
| """若有指定 timeout, 用 asyncio.wait_for 包裝；否則直接 await coro。""" | ||
| if timeout is None: | ||
| return await coro | ||
| return await asyncio.wait_for(coro, timeout=timeout) | ||
|
|
||
| # ========== 讀取操作 ========== | ||
|
|
||
| async def read_coils(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| """讀取線圈狀態 (FC 0x01)""" | ||
| async def read_coils(self, address: int, count: int, unit_id: int = 1, timeout: float | None = None) -> list[bool]: | ||
| """讀取線圈狀態 (FC 0x01) | ||
|
|
||
| Args: | ||
| timeout: 此請求逾時 (秒)；None 沿用 config.timeout | ||
| """ | ||
| client = self._get_client() | ||
| response = await client.read_coils( | ||
| address=address, | ||
| count=count, | ||
| **_slave_kwarg(unit_id), | ||
| response = await self._maybe_wait_for( | ||
| client.read_coils(address=address, count=count, **_slave_kwarg(unit_id)), | ||
| timeout, | ||
| ) | ||
| if response.isError(): | ||
| raise ModbusError(f"讀取線圈失敗: {response}", address=address, unit_id=unit_id, function_code="FC01") | ||
| return list(response.bits[:count]) | ||
|
|
||
| async def read_discrete_inputs(self, address: int, count: int, unit_id: int = 1) -> list[bool]: | ||
| """讀取離散輸入 (FC 0x02)""" | ||
| async def read_discrete_inputs( | ||
| self, address: int, count: int, unit_id: int = 1, timeout: float | None = None | ||
| ) -> list[bool]: | ||
| """讀取離散輸入 (FC 0x02) | ||
|
|
||
| Args: | ||
| timeout: 此請求逾時 (秒)；None 沿用 config.timeout | ||
| """ | ||
| client = self._get_client() | ||
| response = await client.read_discrete_inputs( | ||
| address=address, | ||
| count=count, | ||
| **_slave_kwarg(unit_id), | ||
| response = await self._maybe_wait_for( | ||
| client.read_discrete_inputs(address=address, count=count, **_slave_kwarg(unit_id)), | ||
| timeout, | ||
| ) | ||
| if response.isError(): | ||
| raise ModbusError(f"讀取離散輸入失敗: {response}", address=address, unit_id=unit_id, function_code="FC02") | ||
| return list(response.bits[:count]) | ||
|
|
||
| async def read_holding_registers(self, address: int, count: int, unit_id: int = 1) -> list[int]: | ||
| """讀取保持暫存器 (FC 0x03)""" | ||
| async def read_holding_registers( | ||
| self, address: int, count: int, unit_id: int = 1, timeout: float | None = None | ||
| ) -> list[int]: | ||
| """讀取保持暫存器 (FC 0x03) | ||
|
|
||
| Args: | ||
| timeout: 此請求逾時 (秒)；None 沿用 config.timeout | ||
| """ | ||
| client = self._get_client() | ||
| response = await client.read_holding_registers( | ||
| address=address, | ||
| count=count, | ||
| **_slave_kwarg(unit_id), | ||
| response = await self._maybe_wait_for( | ||
| client.read_holding_registers(address=address, count=count, **_slave_kwarg(unit_id)), | ||
| timeout, | ||
| ) | ||
| if response.isError(): | ||
| raise ModbusError(f"讀取保持暫存器失敗: {response}", address=address, unit_id=unit_id, function_code="FC03") | ||
| return list(response.registers) | ||
|
|
||
| async def read_input_registers(self, address: int, count: int, unit_id: int = 1) -> list[int]: | ||
| """讀取輸入暫存器 (FC 0x04)""" | ||
| async def read_input_registers( | ||
| self, address: int, count: int, unit_id: int = 1, timeout: float | None = None | ||
| ) -> list[int]: | ||
| """讀取輸入暫存器 (FC 0x04) | ||
|
|
||
| Args: | ||
| timeout: 此請求逾時 (秒)；None 沿用 config.timeout | ||
| """ | ||
| client = self._get_client() | ||
| response = await client.read_input_registers( | ||
| address=address, | ||
| count=count, | ||
| **_slave_kwarg(unit_id), | ||
| response = await self._maybe_wait_for( | ||
| client.read_input_registers(address=address, count=count, **_slave_kwarg(unit_id)), | ||
| timeout, | ||
| ) |
…meout 測試 PR #144 Copilot review feedback。 1. AsyncModbusClientBase 抽象基底 + 三個實作 (PymodbusTcpClient / SharedPymodbusTcpClient / PymodbusRtuClient) 的 8 個 read/write 方法, timeout 參數一律改為 keyword-only (`*, timeout: float | None = None`)。 避免 downstream 用 positional 4-arg 形式呼叫,讓未來 signature 演進 不會默默打破 caller。已 grep 確認 csp_lib / tests / docs / examples 皆無 positional 形式使用 timeout,本變更為 signature 收緊 + docstring 標註 keyword-only。 2. 補上 PymodbusTcpClient 的 per-call timeout 單元測試: - timeout 過短時拋 asyncio.TimeoutError 且早返 (< 1.0s) - timeout=None 維持原本行為,呼叫成功返回結果 原本 test_shared_client_timeout.py 只覆蓋 Shared/RTU 的 queue 路徑, 未驗證 PymodbusTcpClient 的 `_maybe_wait_for` 包裝路徑。 3. 新增 TestTimeoutKeywordOnly 守門測試 (parametrize 12 組): 用 inspect.signature 斷言 timeout 參數的 kind 必為 KEYWORD_ONLY, 防止未來有人不小心把 `*` 拿掉。 修改範圍純粹是 PR #144 內的 polish,timeout 參數本身就是此 PR 才首次 加入 (commit abe5a64),尚未隨任何 release 流出,因此 keyword-only 收緊 不算 breaking 歷史 API。 tests/modbus/ 254 passed (含新增 3 個 TCP timeout + 12 個 signature guard)。
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
問題
Closed-loop probe 發現：
SharedPymodbusTcpClient與PymodbusRtuClient各 8 個 read/write method 把timeout寫死成RequestQueueConfig.default_timeout。底層ModbusRequestQueue.submit()早就支援timeout: float | None = None，wrapper 從未把這個參數 plumb 上去。Operator 只能在 queue 層級二選一：改動
所有 read/write method 加
timeout: float | None = Nonekeyword，plumb 進_submit_request → queue.submit(timeout=...)：config.default_timeout（向後相容）容許 operator 對單一慢 unit 開高 timeout（如 PCS 2.5s），其他 fast unit 維持 default 0.8s，CB 真正只在「server 不回」場景才 trip。
純 additive，無 public API 變更語意 → commit type
feat（無!、無 BREAKING CHANGE）→ release-please 推 minor bump。驗證
tests/modbus/test_shared_client_timeout.py12 案（Shared TCP 8 method + RTU 3 method + 1 backward-compat）。Pre-fix FAIL → post-fix 全 PASS。sandbox/modbus_per_call_timeout_demo.py：