From e44285a13ea0fe421b49238f3bdee1b058e4f0cd Mon Sep 17 00:00:00 2001 From: IkukiKanade <734588986@qq.com> Date: Tue, 18 Aug 2026 16:04:00 +0800 Subject: [PATCH 1/4] feat: add conductivity workstation integration mock --- tests/devices/test_conductivity_station.py | 188 ++++++++ .../test_conductivity_station_registry.py | 90 ++++ .../conductivity_station/__init__.py | 5 + .../conductivity_station.json | 29 ++ .../conductivity_station.py | 221 +++++++++ .../conductivity_station/mock_server.py | 380 +++++++++++++++ .../conductivity_station/mock_unilab.py | 448 ++++++++++++++++++ 7 files changed, 1361 insertions(+) create mode 100644 tests/devices/test_conductivity_station.py create mode 100644 tests/devices/test_conductivity_station_registry.py create mode 100644 unilabos/devices/workstation/conductivity_station/__init__.py create mode 100644 unilabos/devices/workstation/conductivity_station/conductivity_station.json create mode 100644 unilabos/devices/workstation/conductivity_station/conductivity_station.py create mode 100644 unilabos/devices/workstation/conductivity_station/mock_server.py create mode 100644 unilabos/devices/workstation/conductivity_station/mock_unilab.py diff --git a/tests/devices/test_conductivity_station.py b/tests/devices/test_conductivity_station.py new file mode 100644 index 000000000..097d198bf --- /dev/null +++ b/tests/devices/test_conductivity_station.py @@ -0,0 +1,188 @@ +"""电导工站 TCP 客户端与模拟服务端集成测试。""" + +from __future__ import annotations + +import json +import math +import socketserver +import threading +import time + +import pytest + +from unilabos.devices.workstation.conductivity_station.conductivity_station import ( + ConductivityStation, + ConductivityStationProtocolError, + ConductivityStationTransportError, +) +from unilabos.devices.workstation.conductivity_station.mock_server import ( + MockConductivityServer, + MockConductivityState, +) + + +@pytest.fixture() +def station() -> ConductivityStation: + state = MockConductivityState(step_interval=0.01) + server = MockConductivityServer(("127.0.0.1", 0), state=state) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + client = ConductivityStation( + ip="127.0.0.1", + port=server.server_address[1], + response_timeout=1.0, + ) + try: + yield client + finally: + client.close() + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_status_material_start_and_result(station: ConductivityStation) -> None: + assert station.station_status()["data"]["robot_arm"] == 1 + assert station.material_status()["data"]["bottle"][:3] == [1, 1, 1] + + started = station.start_batch() + assert started["result"] == 0 + batch_id = started["data"]["batch_id"] + + deadline = time.monotonic() + 2 + while time.monotonic() < deadline: + status = station.batch_status()["data"] + if not status["running"]: + break + time.sleep(0.02) + else: + pytest.fail("模拟批次未在预期时间内完成") + assert status["batch"]["current_test"] == 3 + assert status["batch"]["current_test_step"] == 13 + + result = station.batch_result(batch_id) + tests = result["data"]["batch"]["tests"] + assert [item["state"] for item in tests] == ["finished"] * 3 + assert tests[0]["bottle_code"].startswith("LOT-") + assert all(item["test_time"] for item in tests) + assert tests[0]["R"] > 0 + assert tests[0]["ion_conductivity"] > 0 + assert tests[0]["area"] == pytest.approx(math.pi * 0.5**2) + thickness_cm = (tests[0]["h2"] - tests[0]["h1"]) / 10.0 + expected_ms_cm = thickness_cm / (tests[0]["R"] * tests[0]["area"]) * 1000 + assert tests[0]["ion_conductivity"] == pytest.approx(expected_ms_cm, rel=1e-5) + + +def test_manual_validation_and_stop_after_current_sample( + station: ConductivityStation, +) -> None: + with pytest.raises(ValueError, match="1 到 13"): + station.manual_run(14) + + batch_id = station.start_batch()["data"]["batch_id"] + assert station.manual_run(6)["result"] == 0 + assert station.stop_current_batch()["result"] == 0 + + deadline = time.monotonic() + 1 + while time.monotonic() < deadline: + status = station.batch_status()["data"] + if not status["running"]: + break + time.sleep(0.02) + assert status["batch"]["state"] == "stopped" + tests = station.batch_result(batch_id)["data"]["batch"]["tests"] + assert tests[0]["state"] == "finished" + assert tests[1]["state"] == "not_started" + + +def test_unknown_batch_and_clear(station: ConductivityStation) -> None: + assert station.batch_result("missing")["result"] == 9 + assert station.clear_current_batch()["result"] == 0 + status = station.batch_status()["data"] + assert status == {"running": False, "batch": None} + + +def _scripted_station(plans: list[object]) -> tuple[ConductivityStation, socketserver.TCPServer, threading.Thread]: + """启动一个按连接顺序返回分片、断线或指定响应的协议测试服务。""" + + class Handler(socketserver.BaseRequestHandler): + def handle(self) -> None: + received = bytearray() + while b"\r\n" not in received: + chunk = self.request.recv(4096) + if not chunk: + return + received.extend(chunk) + request = json.loads(bytes(received).split(b"\r\n", 1)[0]) + with self.server.plan_lock: # type: ignore[attr-defined] + plan = self.server.plans.pop(0) # type: ignore[attr-defined] + if plan is None: + return + if callable(plan): + plan = plan(request) + chunks = plan if isinstance(plan, list) else [plan] + for chunk in chunks: + self.request.sendall(chunk) + + server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler) + server.plans = list(plans) # type: ignore[attr-defined] + server.plan_lock = threading.Lock() # type: ignore[attr-defined] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + client = ConductivityStation( + ip="127.0.0.1", port=server.server_address[1], response_timeout=0.5 + ) + return client, server, thread + + +def _close_scripted_station( + client: ConductivityStation, + server: socketserver.TCPServer, + thread: threading.Thread, +) -> None: + client.close() + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_tcp_fragmentation_and_read_retry() -> None: + def fragmented(request: dict) -> list[bytes]: + payload = json.dumps( + {"request_id": request["request_id"], "result": 0, "data": {"running": False, "batch": None}} + ).encode() + return [payload[:8], payload[8:] + b"\r\n"] + + client, server, thread = _scripted_station([None, fragmented]) + try: + # 查询动作首次断线后可安全重连，且能拼接分片响应。 + assert client.batch_status()["data"]["running"] is False + assert server.plans == [] # type: ignore[attr-defined] + finally: + _close_scripted_station(client, server, thread) + +@pytest.mark.parametrize( + "response,exception", + [ + (b'{"request_id":999,"result":0}\r\n', ConductivityStationProtocolError), + (b'{not-json}\r\n', ConductivityStationProtocolError), + ], +) +def test_invalid_responses_are_rejected(response: bytes, exception: type[Exception]) -> None: + client, server, thread = _scripted_station([response]) + try: + with pytest.raises(exception): + client.start_batch() + finally: + _close_scripted_station(client, server, thread) + + +def test_side_effect_action_is_not_retried_after_disconnect() -> None: + client, server, thread = _scripted_station([None, b'{"request_id":1,"result":0}\r\n']) + try: + with pytest.raises(ConductivityStationTransportError): + client.start_batch() + # 第二个计划仍在，证明有副作用动作未被自动重发。 + assert len(server.plans) == 1 # type: ignore[attr-defined] + finally: + _close_scripted_station(client, server, thread) diff --git a/tests/devices/test_conductivity_station_registry.py b/tests/devices/test_conductivity_station_registry.py new file mode 100644 index 000000000..7ee04ee48 --- /dev/null +++ b/tests/devices/test_conductivity_station_registry.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import ast +import json +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +DEVICE_DIR = ( + ROOT + / "unilabos" + / "devices" + / "workstation" + / "conductivity_station" +) + + +def test_conductivity_station_actions_are_discoverable_by_registry() -> None: + tree = ast.parse( + (DEVICE_DIR / "conductivity_station.py").read_text(encoding="utf-8") + ) + device_class = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "ConductivityStation" + ) + device_decorator = next( + decorator + for decorator in device_class.decorator_list + if isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Name) + and decorator.func.id == "device" + ) + device_id = next( + keyword.value.value + for keyword in device_decorator.keywords + if keyword.arg == "id" and isinstance(keyword.value, ast.Constant) + ) + actions = { + method.name + for method in device_class.body + if isinstance(method, ast.FunctionDef) + and any( + (isinstance(decorator, ast.Name) and decorator.id == "action") + or ( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Name) + and decorator.func.id == "action" + ) + for decorator in method.decorator_list + ) + } + init_method = next( + method + for method in device_class.body + if isinstance(method, ast.FunctionDef) and method.name == "__init__" + ) + + assert device_id == "conductivity_station" + assert actions == { + "station_status", + "material_status", + "batch_status", + "batch_result", + "start_batch", + "stop_current_batch", + "manual_run", + "clear_current_batch", + } + assert {arg.arg for arg in init_method.args.args if arg.arg != "self"} == { + "ip", + "port", + "connect_timeout", + "response_timeout", + "max_message_bytes", + "encoding", + "frame_delimiter", + "station_action_names", + } + + +def test_device_graph_references_registered_device() -> None: + graph = json.loads( + (DEVICE_DIR / "conductivity_station.json").read_text(encoding="utf-8") + ) + + assert graph["links"] == [] + assert graph["nodes"][0]["id"] == "CONDUCTIVITY_STATION" + assert graph["nodes"][0]["class"] == "conductivity_station" + assert graph["nodes"][0]["config"]["port"] == 19091 diff --git a/unilabos/devices/workstation/conductivity_station/__init__.py b/unilabos/devices/workstation/conductivity_station/__init__.py new file mode 100644 index 000000000..ef00f5920 --- /dev/null +++ b/unilabos/devices/workstation/conductivity_station/__init__.py @@ -0,0 +1,5 @@ +"""电导率自动化工站 TCP 设备。""" + +from .conductivity_station import ConductivityStation + +__all__ = ["ConductivityStation"] diff --git a/unilabos/devices/workstation/conductivity_station/conductivity_station.json b/unilabos/devices/workstation/conductivity_station/conductivity_station.json new file mode 100644 index 000000000..6d4194330 --- /dev/null +++ b/unilabos/devices/workstation/conductivity_station/conductivity_station.json @@ -0,0 +1,29 @@ +{ + "nodes": [ + { + "id": "CONDUCTIVITY_STATION", + "name": "电导率自动化测试工站", + "parent": null, + "type": "device", + "class": "conductivity_station", + "position": { + "x": 720.0, + "y": 200.0, + "z": 0 + }, + "config": { + "ip": "127.0.0.1", + "port": 19091, + "connect_timeout": 5.0, + "response_timeout": 10.0, + "max_message_bytes": 4194304, + "encoding": "utf-8", + "frame_delimiter": "\\r\\n", + "station_action_names": {} + }, + "data": {}, + "children": [] + } + ], + "links": [] +} diff --git a/unilabos/devices/workstation/conductivity_station/conductivity_station.py b/unilabos/devices/workstation/conductivity_station/conductivity_station.py new file mode 100644 index 000000000..47acae374 --- /dev/null +++ b/unilabos/devices/workstation/conductivity_station/conductivity_station.py @@ -0,0 +1,221 @@ +"""电导率自动化工站 TCP 客户端及 UniLab 动作。 + +合作方工站作为 TCP 服务端，UniLab 作为客户端。报文为 UTF-8 JSON，使用 +CRLF 分帧。查询类动作允许在连接中断后重连重试一次；启动、停止和手动运行 +属于有副作用的命令，发送结果不明确时不会自动重发，避免重复执行。 +""" + +from __future__ import annotations + +import json +import socket +import threading +from itertools import count +from typing import Any + +from unilabos.registry.decorators import action, device + + +class ConductivityStationTransportError(RuntimeError): + """TCP 连接或报文传输失败。""" + + +class ConductivityStationProtocolError(RuntimeError): + """服务端响应不符合电导工站协议。""" + + +@device( + id="conductivity_station", + category=["workstation", "conductivity"], + displayname="电导率自动化测试工站", + description="通过 TCP JSON/CRLF 协议控制电导率自动化测试工站", + version="1.0.0", +) +class ConductivityStation: + """电导工站长连接客户端。""" + + def __init__( + self, + ip: str = "127.0.0.1", + port: int = 19091, + connect_timeout: float = 5.0, + response_timeout: float = 10.0, + max_message_bytes: int = 4194304, + encoding: str = "utf-8", + frame_delimiter: str = "\\r\\n", + station_action_names: dict[str, str] | None = None, + **_: Any, + ) -> None: + self.ip = str(ip) + self.port = int(port) + self.connect_timeout = float(connect_timeout) + self.response_timeout = float(response_timeout) + self.max_message_bytes = int(max_message_bytes) + self.encoding = str(encoding).strip() or "utf-8" + delimiter_text = ( + str(frame_delimiter).replace("\\r", "\r").replace("\\n", "\n") + ) + self.frame_delimiter = delimiter_text.encode(self.encoding) + if not self.frame_delimiter: + raise ValueError("frame_delimiter 不能为空") + self.station_action_names = { + str(key): str(value) + for key, value in (station_action_names or {}).items() + if str(key) and str(value) + } + self.status = "idle" + self._sock: socket.socket | None = None + self._recv_buffer = bytearray() + self._request_ids = count(1) + self._lock = threading.RLock() + + def close(self) -> None: + """关闭当前 TCP 连接。""" + with self._lock: + self._disconnect() + + def _disconnect(self) -> None: + sock, self._sock = self._sock, None + self._recv_buffer.clear() + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + sock.close() + except OSError: + pass + + def _connect(self) -> socket.socket: + if self._sock is None: + try: + sock = socket.create_connection( + (self.ip, self.port), timeout=self.connect_timeout + ) + sock.settimeout(self.response_timeout) + self._sock = sock + except OSError as exc: + self._disconnect() + raise ConductivityStationTransportError( + f"无法连接电导工站 {self.ip}:{self.port}: {exc}" + ) from exc + return self._sock + + def _read_frame(self, sock: socket.socket) -> bytes: + delimiter = self.frame_delimiter + while True: + marker = self._recv_buffer.find(delimiter) + if marker >= 0: + frame = bytes(self._recv_buffer[:marker]) + del self._recv_buffer[: marker + len(delimiter)] + return frame + if len(self._recv_buffer) > self.max_message_bytes: + raise ConductivityStationProtocolError( + f"响应超过最大长度 {self.max_message_bytes} 字节" + ) + try: + chunk = sock.recv(65536) + except (OSError, socket.timeout) as exc: + raise ConductivityStationTransportError( + f"等待电导工站响应失败: {exc}" + ) from exc + if not chunk: + raise ConductivityStationTransportError("电导工站在响应前关闭连接") + self._recv_buffer.extend(chunk) + + def _exchange(self, request: dict[str, Any]) -> dict[str, Any]: + sock = self._connect() + encoded = json.dumps( + request, ensure_ascii=False, separators=(",", ":") + ).encode(self.encoding) + self.frame_delimiter + if len(encoded) > self.max_message_bytes: + raise ConductivityStationProtocolError( + f"请求超过最大长度 {self.max_message_bytes} 字节" + ) + try: + sock.sendall(encoded) + frame = self._read_frame(sock) + except Exception: + self._disconnect() + raise + try: + response = json.loads(frame.decode(self.encoding)) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + self._disconnect() + raise ConductivityStationProtocolError( + f"响应不是有效 {self.encoding} JSON: {exc}" + ) from exc + if not isinstance(response, dict): + raise ConductivityStationProtocolError("响应必须是 JSON 对象") + if response.get("request_id") != request["request_id"]: + raise ConductivityStationProtocolError( + "响应 request_id 不匹配: " + f"期望 {request['request_id']}，实际 {response.get('request_id')}" + ) + if "result" not in response: + raise ConductivityStationProtocolError("响应缺少 result 字段") + return response + + def _request( + self, + action_name: str, + param: dict[str, Any] | None = None, + *, + retry_read: bool = False, + ) -> dict[str, Any]: + with self._lock: + request: dict[str, Any] = { + "request_id": next(self._request_ids), + "action": self.station_action_names.get(action_name, action_name), + } + if param is not None: + request["param"] = param + attempts = 2 if retry_read else 1 + for attempt in range(attempts): + try: + return self._exchange(request) + except ConductivityStationTransportError: + if attempt + 1 >= attempts: + raise + self._disconnect() + raise AssertionError("unreachable") + + @action(always_free=True, description="查询电导工站各机构在线状态") + def station_status(self) -> dict[str, Any]: + return self._request("station_status", retry_read=True) + + @action(always_free=True, description="查询料瓶、模具和漏斗占位情况") + def material_status(self) -> dict[str, Any]: + return self._request("material_status", retry_read=True) + + @action(always_free=True, description="查询当前电导测试批次和执行进度") + def batch_status(self) -> dict[str, Any]: + return self._request("batch_status", retry_read=True) + + @action(always_free=True, description="按批次号查询样品测试结果") + def batch_result(self, batch_id: str) -> dict[str, Any]: + if not str(batch_id).strip(): + raise ValueError("batch_id 不能为空") + return self._request( + "batch_result", {"batch_id": str(batch_id).strip()}, retry_read=True + ) + + @action(description="上料完成后启动整批自动测试") + def start_batch(self) -> dict[str, Any]: + return self._request("start_batch") + + @action(description="当前样品完成后停止整批任务") + def stop_current_batch(self) -> dict[str, Any]: + return self._request("stop_current_batch") + + @action(description="切换手动模式并单独执行指定步骤") + def manual_run(self, step: int) -> dict[str, Any]: + step_value = int(step) + if not 1 <= step_value <= 13: + raise ValueError("step 必须在 1 到 13 之间") + return self._request("manual_run", {"step": step_value}) + + @action(description="清理非运行状态下的当前批次（动作名待合作方最终确认）") + def clear_current_batch(self) -> dict[str, Any]: + return self._request("clear_current_batch") diff --git a/unilabos/devices/workstation/conductivity_station/mock_server.py b/unilabos/devices/workstation/conductivity_station/mock_server.py new file mode 100644 index 000000000..6ff359890 --- /dev/null +++ b/unilabos/devices/workstation/conductivity_station/mock_server.py @@ -0,0 +1,380 @@ +"""电导工站 TCP 协议模拟服务端。""" + +from __future__ import annotations + +import argparse +import copy +import json +import math +import socketserver +import threading +import time +from datetime import datetime +from typing import Any + + +STEP_NAMES = { + 1: "模具拆解", + 2: "模具测厚", + 3: "模具转移", + 4: "漏斗转移", + 5: "烧结料瓶扫码与转移", + 6: "加粉", + 7: "烧结料瓶暂存", + 8: "模具组装", + 9: "模具转移", + 10: "EIS测试", + 11: "模具测厚", + 12: "电子电导率测试", + 13: "测试物料转移至托盘", +} + +# 合作方高保真模型确认使用直径 1 cm 圆片：A = π × (0.5 cm)²。 +DISC_AREA_CM2 = math.pi * 0.5**2 + + +class MockConductivityState: + """线程安全的批次模拟状态。""" + + def __init__( + self, + *, + step_interval: float = 0.1, + failure_sample: int | None = None, + failure_step: int = 10, + ) -> None: + self.step_interval = max(0.01, float(step_interval)) + self.failure_sample = failure_sample + self.failure_step = int(failure_step) + self.lock = threading.RLock() + self.started_monotonic: float | None = None + self.stop_requested = False + self.stop_after_sample: int | None = None + self.batch: dict[str, Any] | None = None + self.history: dict[str, dict[str, Any]] = {} + self.materials = { + "bottle": [1, 1, 1, 0, 0, 0, 0, 0, 0, 0], + "mold": [0, 0, 0, 1, 1, 1, 0, 0, 0, 0], + "funnel": [0, 0, 0, 0, 0, 0, 1, 1, 1, 0], + } + + @staticmethod + def _positions(values: list[int]) -> list[int]: + return [index for index, occupied in enumerate(values, start=1) if occupied] + + def _new_test(self, index: int, bottle: int, mold: int, funnel: int) -> dict[str, Any]: + return { + "state": "not_started", + "test_time": "", + "bottle": bottle, + "mold": mold, + "funnel": funnel, + "step": 0, + "bottle_code": "", + "recipe": "", + "formula": "", + "temperature": 0.0, + "due_pressure": 3, + "pressure_time": 10, + "R": 0.0, + "h1": 0.0, + "h2": 0.0, + "area": DISC_AREA_CM2, + "ion_conductivity": 0.0, + "elec_conductivity_tested": False, + "elec_R": 0.0, + "elec_conductivity": 0.0, + "_sample_index": index, + } + + def _update(self) -> None: + if not self.batch or self.started_monotonic is None or not self.batch["running"]: + return + tests = self.batch["tests"] + elapsed_steps = int((time.monotonic() - self.started_monotonic) / self.step_interval) + target_completed = elapsed_steps // 13 + active_step = elapsed_steps % 13 + 1 + + if self.stop_after_sample is not None: + target_completed = min(target_completed, self.stop_after_sample) + + for index, test in enumerate(tests, start=1): + if test["state"] in {"finished", "failed"}: + continue + failure_reached = ( + index == self.failure_sample + and elapsed_steps >= (index - 1) * 13 + self.failure_step - 1 + ) + if failure_reached: + if not test["test_time"]: + test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + test["state"] = "failed" + test["step"] = self.failure_step + test["error_code"] = "MOCK_STEP_FAILED" + test["error_message"] = f"模拟失败：{STEP_NAMES[self.failure_step]}" + self.batch["running"] = False + self.batch["state"] = "failed" + break + if index <= target_completed: + self._finish_test(test) + elif index == target_completed + 1 and not ( + self.stop_after_sample is not None + and index > self.stop_after_sample + ): + if not test["test_time"]: + test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + test["state"] = "in_progress" + test["step"] = active_step + break + + finished = sum(item["state"] == "finished" for item in tests) + failed = sum(item["state"] == "failed" for item in tests) + self.batch["finished_count"] = finished + self.batch["failed_count"] = failed + active = next( + (item for item in tests if item["state"] == "in_progress"), None + ) + if active: + self.batch["current_test"] = active["_sample_index"] + self.batch["current_test_step"] = active["step"] + if ( + self.stop_after_sample is not None + and finished >= self.stop_after_sample + ): + self.batch["running"] = False + self.batch["state"] = "stopped" + elif finished + failed == len(tests): + self.batch["running"] = False + self.batch["state"] = "completed" if failed == 0 else "failed" + self.batch["current_test"] = len(tests) + self.batch["current_test_step"] = tests[-1]["step"] if tests else 0 + self.history[self.batch["batch_id"]] = copy.deepcopy(self.batch) + + @staticmethod + def _finish_test(test: dict[str, Any]) -> None: + index = int(test["_sample_index"]) + if not test["test_time"]: + test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + resistance_ohm = round(80.0 + index * 4.25, 3) + h1_mm = 8.0 + h2_mm = round(8.8 + index * 0.02, 3) + thickness_cm = (h2_mm - h1_mm) / 10.0 + ionic_conductivity_ms_cm = ( + thickness_cm / (resistance_ohm * DISC_AREA_CM2) * 1000.0 + ) + test.update( + { + "state": "finished", + "step": 13, + "bottle_code": f"LOT-260812-{index:03d}-CRU-S-{index:03d}-260812", + "recipe": "R-202608-001", + "formula": "Li6PS5Cl", + "temperature": round(26.5 + index * 0.1, 2), + "R": resistance_ohm, + "h1": h1_mm, + "h2": h2_mm, + "area": DISC_AREA_CM2, + "ion_conductivity": round(ionic_conductivity_ms_cm, 6), + } + ) + + @staticmethod + def _public(value: Any) -> Any: + if isinstance(value, dict): + return { + key: MockConductivityState._public(item) + for key, item in value.items() + if not key.startswith("_") + } + if isinstance(value, list): + return [MockConductivityState._public(item) for item in value] + return value + + def handle(self, request: dict[str, Any]) -> dict[str, Any]: + request_id = request.get("request_id") + action = request.get("action") + param = request.get("param") or {} + with self.lock: + self._update() + if action == "station_status": + return self._response( + request_id, + data={ + "robot_arm": 1, + "scanner": 1, + "lid_open_close_mechanism": 1, + "powder_adding_mechanism": 1, + "tablet_pressing_mechanism": 1, + "electrochemical_workstation": 1, + "stack_rack": 1, + }, + ) + if action == "material_status": + return self._response(request_id, data=copy.deepcopy(self.materials)) + if action == "batch_status": + if not self.batch: + return self._response(request_id, data={"running": False, "batch": None}) + summary_keys = ( + "batch_id", + "test_round", + "bottle", + "mold", + "funnel", + "current_test", + "current_test_step", + "finished_count", + "failed_count", + "state", + ) + summary = {key: self.batch.get(key) for key in summary_keys} + return self._response( + request_id, + data={"running": self.batch["running"], "batch": summary}, + ) + if action == "batch_result": + batch_id = str(param.get("batch_id") or "") + batch = self.history.get(batch_id) + if not batch: + return self._response(request_id, result=9) + return self._response( + request_id, + data={ + "batch": { + "batch_id": batch_id, + "test_round": batch["test_round"], + "state": batch["state"], + "tests": self._public(copy.deepcopy(batch["tests"])), + } + }, + ) + if action == "start_batch": + if self.batch and self.batch["running"]: + return self._response(request_id, result=4) + positions = [self._positions(self.materials[key]) for key in ("bottle", "mold", "funnel")] + if not positions[0] or len({len(items) for items in positions}) != 1: + return self._response(request_id, result=2) + batch_id = datetime.now().strftime("%Y%m%d%H%M%S%f")[:17] + tests = [ + self._new_test(index, bottle, mold, funnel) + for index, (bottle, mold, funnel) in enumerate(zip(*positions), start=1) + ] + self.batch = { + "batch_id": batch_id, + "test_round": len(tests), + "bottle": positions[0], + "mold": positions[1], + "funnel": positions[2], + "current_test": 1, + "current_test_step": 1, + "finished_count": 0, + "failed_count": 0, + "running": True, + "state": "running", + "tests": tests, + } + self.started_monotonic = time.monotonic() + self.stop_requested = False + self.stop_after_sample = None + self.history[batch_id] = copy.deepcopy(self.batch) + return self._response(request_id, data={"batch_id": batch_id}) + if action == "stop_current_batch": + if not self.batch or not self.batch["running"]: + return self._response(request_id, result=3) + self.stop_requested = True + self.stop_after_sample = int(self.batch.get("current_test") or 1) + self.batch["state"] = "stopping" + return self._response(request_id) + if action == "manual_run": + if not self.batch or not self.batch["running"]: + return self._response(request_id, result=3) + step = int(param.get("step") or 0) + if not 1 <= step <= 13: + return self._response(request_id, result=5) + active = next( + (item for item in self.batch["tests"] if item["state"] == "in_progress"), + self.batch["tests"][0], + ) + active["state"] = "in_progress" + active["step"] = step + if not active["test_time"]: + active["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + return self._response( + request_id, + data={ + "batch_id": self.batch["batch_id"], + "test_time": active["test_time"], + }, + ) + if action == "clear_current_batch": + if self.batch and self.batch["running"]: + return self._response(request_id, result=4) + self.batch = None + self.started_monotonic = None + self.stop_requested = False + self.stop_after_sample = None + return self._response(request_id) + return self._response(request_id, result=8) + + @staticmethod + def _response( + request_id: Any, result: int = 0, data: dict[str, Any] | None = None + ) -> dict[str, Any]: + response: dict[str, Any] = {"request_id": request_id, "result": result} + if data is not None: + response["data"] = data + return response + + +class _Handler(socketserver.StreamRequestHandler): + def handle(self) -> None: + state: MockConductivityState = self.server.state # type: ignore[attr-defined] + while True: + raw = self.rfile.readline() + if not raw: + return + try: + request = json.loads(raw.decode("utf-8").strip()) + if not isinstance(request, dict): + raise ValueError("请求必须是 JSON 对象") + response = state.handle(request) + except Exception as exc: # noqa: BLE001 + response = {"request_id": None, "result": 7, "message": str(exc)} + payload = json.dumps(response, ensure_ascii=False, separators=(",", ":")) + self.wfile.write((payload + "\r\n").encode("utf-8")) + self.wfile.flush() + + +class MockConductivityServer(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + def __init__( + self, + address: tuple[str, int] = ("127.0.0.1", 19091), + *, + state: MockConductivityState | None = None, + ) -> None: + self.state = state or MockConductivityState() + super().__init__(address, _Handler) + + +def main() -> None: + parser = argparse.ArgumentParser(description="电导工站 TCP 协议模拟服务端") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=19091) + parser.add_argument("--step-interval", type=float, default=0.5) + parser.add_argument("--failure-sample", type=int) + parser.add_argument("--failure-step", type=int, default=10) + args = parser.parse_args() + state = MockConductivityState( + step_interval=args.step_interval, + failure_sample=args.failure_sample, + failure_step=args.failure_step, + ) + with MockConductivityServer((args.host, args.port), state=state) as server: + print(f"模拟电导工站监听 {args.host}:{server.server_address[1]}", flush=True) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/unilabos/devices/workstation/conductivity_station/mock_unilab.py b/unilabos/devices/workstation/conductivity_station/mock_unilab.py new file mode 100644 index 000000000..98d82bef4 --- /dev/null +++ b/unilabos/devices/workstation/conductivity_station/mock_unilab.py @@ -0,0 +1,448 @@ +"""用于电导工站联调的最小 UniLab Job API 模拟网关。 + +该网关只用于没有编译 ``unilabos_msgs`` 的本地开发环境。平台仍使用正式的 +UniLab Job API 契约；网关把 Job 动作分发到 ``ConductivityStation`` 注册设备， +设备再通过真实 TCP/CRLF 协议访问模拟工站。 +""" + +from __future__ import annotations + +import argparse +import html +import inspect +import json +import threading +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import urlparse + +from .conductivity_station import ConductivityStation +from .mock_server import MockConductivityServer, MockConductivityState + + +class MockUniLabJobState: + """线程安全的 Job 存储与动作分发器。""" + + def __init__( + self, + device: ConductivityStation, + device_id: str = "CONDUCTIVITY_STATION", + ) -> None: + self.device = device + self.device_id = device_id + self.jobs: dict[str, dict[str, Any]] = {} + self.lock = threading.RLock() + + @staticmethod + def _json_type(annotation: Any) -> str: + """将动作参数注解转换为页面与 API 使用的简化 JSON Schema 类型。""" + text = str(annotation).lower() + if annotation is int or text in {"int", ""}: + return "integer" + if annotation is float or text in {"float", ""}: + return "number" + if annotation is bool or text in {"bool", ""}: + return "boolean" + if "list" in text or "tuple" in text or "set" in text: + return "array" + if "dict" in text: + return "object" + return "string" + + def actions(self) -> dict[str, dict[str, Any]]: + """从实际设备类的 ``@action`` 元数据动态生成动作目录。""" + catalog: dict[str, dict[str, Any]] = {} + for name, method in inspect.getmembers(type(self.device), inspect.isfunction): + meta = getattr(method, "_action_registry_meta", None) + if not isinstance(meta, dict): + continue + signature = inspect.signature(method) + properties: dict[str, dict[str, Any]] = {} + required: list[str] = [] + parameters: list[dict[str, Any]] = [] + for parameter in signature.parameters.values(): + if parameter.name == "self": + continue + parameter_type = self._json_type(parameter.annotation) + is_required = parameter.default is inspect.Parameter.empty + item: dict[str, Any] = { + "name": parameter.name, + "type": parameter_type, + "required": is_required, + } + schema_item: dict[str, Any] = {"type": parameter_type} + if is_required: + required.append(parameter.name) + else: + item["default"] = parameter.default + schema_item["default"] = parameter.default + parameters.append(item) + properties[parameter.name] = schema_item + schema: dict[str, Any] = { + "type": "object", + "properties": properties, + "additionalProperties": False, + } + if required: + schema["required"] = required + catalog[name] = { + "name": name, + "description": str(meta.get("description") or ""), + "always_free": bool(meta.get("always_free")), + "parameters": parameters, + "schema": schema, + } + return dict(sorted(catalog.items())) + + def submit( + self, device_id: str, action_name: str, action_args: dict[str, Any] + ) -> dict[str, Any]: + job_id = str(uuid.uuid4()) + if device_id != self.device_id: + result = { + "jobId": job_id, + "status": 6, + "result": {"error": f"Device not found: {device_id}"}, + } + with self.lock: + self.jobs[job_id] = result + return result + method = getattr(self.device, action_name, None) + if action_name not in self.actions() or method is None: + result = { + "jobId": job_id, + "status": 6, + "result": {"error": f"Action not found: {action_name}"}, + } + with self.lock: + self.jobs[job_id] = result + return result + try: + return_value = method(**action_args) + result = { + "jobId": job_id, + "status": 4, + "result": {"return_value": return_value}, + } + except Exception as exc: # noqa: BLE001 + result = { + "jobId": job_id, + "status": 6, + "result": {"error": str(exc)}, + } + with self.lock: + self.jobs[job_id] = result + return result + + def status(self, job_id: str) -> dict[str, Any]: + with self.lock: + return self.jobs.get( + job_id, + { + "jobId": job_id, + "status": 6, + "result": {"error": "Job not found"}, + }, + ) + + +class _JobApiHandler(BaseHTTPRequestHandler): + server_version = "MockUniLab/1.0" + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + return + + @property + def state(self) -> MockUniLabJobState: + return self.server.state # type: ignore[attr-defined,no-any-return] + + def _json_body(self) -> dict[str, Any]: + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length) if length else b"{}" + value = json.loads(raw.decode("utf-8")) + if not isinstance(value, dict): + raise ValueError("请求体必须是 JSON 对象") + return value + + def _send(self, data: Any, *, code: int = 0, message: str = "success") -> None: + payload = json.dumps( + {"code": code, "data": data, "message": message}, + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _send_html(self, content: str) -> None: + payload = content.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _status_page(self) -> str: + action_rows = [] + for action in self.state.actions().values(): + parameters = ", ".join( + f"{item['name']}: {item['type']}" + for item in action["parameters"] + ) or "无" + mode = "只读/免排队" if action["always_free"] else "控制动作" + action_rows.append( + "" + f"{html.escape(action['name'])}" + f"{html.escape(parameters)}" + f"{html.escape(action['description'])}" + f"{mode}" + "" + ) + rows = "".join(action_rows) + device_id = html.escape(self.state.device_id) + return f""" + + + + + Mock UniLab 主机状态 + + + +
+

UniLab 主机状态 · Mock

+

电导工站本地接口模拟与动作注册展示

+
+
模拟环境：此页面不是完整 UniLab/ROS 运行时，仅用于验证驱动动作发现和 Job API 联调。
+
+
+
{device_id}Onlinelocal-mock
+

设备类：conductivity_station　动作数量：{len(action_rows)}

+
+
+

已发现的注册动作

+ + + {rows} +
动作参数说明模式
+
+
+ 动作 API： + /api/v1/devices/{device_id}/actions +
+
+ +""" + + def do_POST(self) -> None: # noqa: N802 + if urlparse(self.path).path != "/api/v1/job/add": + self._send({}, code=1, message="not found") + return + try: + body = self._json_body() + device_id = str(body.get("device_id") or "") + action_name = str(body.get("action") or "") + action_args = body.get("action_args") or {} + if not isinstance(action_args, dict): + raise ValueError("action_args 必须是对象") + job = self.state.submit(device_id, action_name, action_args) + self._send({"jobId": job["jobId"], "status": 1}) + except Exception as exc: # noqa: BLE001 + self._send({}, code=1, message=str(exc)) + + def do_GET(self) -> None: # noqa: N802 + path = urlparse(self.path).path + if path in {"/", "/status", "/registry-editor"}: + self._send_html(self._status_page()) + return + if path == "/api/v1/online-devices": + self._send( + { + "online_devices": { + self.state.device_id: { + "device_key": f"/devices/{self.state.device_id}", + "namespace": "/devices", + "machine_name": "local-mock", + } + }, + "total_count": 1, + } + ) + return + action_prefix = f"/api/v1/devices/{self.state.device_id}/actions" + if path == action_prefix: + actions = self.state.actions() + self._send( + { + "device_id": self.state.device_id, + "actions": actions, + "total_count": len(actions), + "environment": "mock", + } + ) + return + schema_marker = action_prefix + "/" + schema_suffix = "/schema" + if path.startswith(schema_marker) and path.endswith(schema_suffix): + action_name = path[len(schema_marker) : -len(schema_suffix)] + action = self.state.actions().get(action_name) + if action is None: + self._send({}, code=1, message=f"Action not found: {action_name}") + else: + self._send( + { + "device_id": self.state.device_id, + "action": action_name, + "schema": action["schema"], + } + ) + return + if path == "/api/v1/actions": + actions = self.state.actions() + self._send( + { + "devices": {self.state.device_id: actions}, + "total_count": len(actions), + "environment": "mock", + } + ) + return + prefix, suffix = "/api/v1/job/", "/status" + if path.startswith(prefix) and path.endswith(suffix): + job_id = path[len(prefix) : -len(suffix)] + self._send(self.state.status(job_id)) + return + self._send({}, code=1, message="not found") + + +class MockUniLabJobServer(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + def __init__( + self, + address: tuple[str, int], + *, + state: MockUniLabJobState, + ) -> None: + self.state = state + super().__init__(address, _JobApiHandler) + + +class ConductivityIntegrationMock: + """同时管理模拟工站和模拟 UniLab Job API，供测试及演示使用。""" + + def __init__( + self, + host: str = "127.0.0.1", + api_port: int = 18002, + station_port: int = 19091, + step_interval: float = 0.1, + failure_sample: int | None = None, + failure_step: int = 10, + ) -> None: + station_state = MockConductivityState( + step_interval=step_interval, + failure_sample=failure_sample, + failure_step=failure_step, + ) + self.station_server = MockConductivityServer( + (host, station_port), state=station_state + ) + actual_station_port = int(self.station_server.server_address[1]) + self.device = ConductivityStation(ip=host, port=actual_station_port) + self.job_state = MockUniLabJobState(self.device) + self.api_server = MockUniLabJobServer((host, api_port), state=self.job_state) + self._threads: list[threading.Thread] = [] + + @property + def api_port(self) -> int: + return int(self.api_server.server_address[1]) + + @property + def station_port(self) -> int: + return int(self.station_server.server_address[1]) + + def start(self) -> None: + for server, name in ( + (self.station_server, "mock-conductivity-station"), + (self.api_server, "mock-unilab-job-api"), + ): + thread = threading.Thread( + target=server.serve_forever, name=name, daemon=True + ) + thread.start() + self._threads.append(thread) + + def close(self) -> None: + self.device.close() + for server in (self.api_server, self.station_server): + server.shutdown() + server.server_close() + for thread in self._threads: + thread.join(timeout=2) + self._threads.clear() + + def __enter__(self) -> "ConductivityIntegrationMock": + self.start() + return self + + def __exit__(self, *_: Any) -> None: + self.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="电导工站 UniLab 全链路模拟器") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--api-port", type=int, default=18002) + parser.add_argument("--station-port", type=int, default=19091) + parser.add_argument("--step-interval", type=float, default=0.5) + parser.add_argument("--failure-sample", type=int) + parser.add_argument("--failure-step", type=int, default=10) + args = parser.parse_args() + mock = ConductivityIntegrationMock( + host=args.host, + api_port=args.api_port, + station_port=args.station_port, + step_interval=args.step_interval, + failure_sample=args.failure_sample, + failure_step=args.failure_step, + ) + mock.start() + print( + f"模拟 UniLab Job API: http://{args.host}:{mock.api_port}/api/v1\n" + f"模拟电导工站 TCP: {args.host}:{mock.station_port}", + flush=True, + ) + try: + threading.Event().wait() + except KeyboardInterrupt: + pass + finally: + mock.close() + + +if __name__ == "__main__": + main() From d2f421cc8c2ec2b3ab36e40c6694dc3232026834 Mon Sep 17 00:00:00 2001 From: IkukiKanade <734588986@qq.com> Date: Tue, 18 Aug 2026 17:06:38 +0800 Subject: [PATCH 2/4] chore: remove deployment graph from conductivity driver --- .../test_conductivity_station_registry.py | 12 -------- .../conductivity_station.json | 29 ------------------- 2 files changed, 41 deletions(-) delete mode 100644 unilabos/devices/workstation/conductivity_station/conductivity_station.json diff --git a/tests/devices/test_conductivity_station_registry.py b/tests/devices/test_conductivity_station_registry.py index 7ee04ee48..ca6cb8cb7 100644 --- a/tests/devices/test_conductivity_station_registry.py +++ b/tests/devices/test_conductivity_station_registry.py @@ -1,7 +1,6 @@ from __future__ import annotations import ast -import json from pathlib import Path @@ -77,14 +76,3 @@ def test_conductivity_station_actions_are_discoverable_by_registry() -> None: "frame_delimiter", "station_action_names", } - - -def test_device_graph_references_registered_device() -> None: - graph = json.loads( - (DEVICE_DIR / "conductivity_station.json").read_text(encoding="utf-8") - ) - - assert graph["links"] == [] - assert graph["nodes"][0]["id"] == "CONDUCTIVITY_STATION" - assert graph["nodes"][0]["class"] == "conductivity_station" - assert graph["nodes"][0]["config"]["port"] == 19091 diff --git a/unilabos/devices/workstation/conductivity_station/conductivity_station.json b/unilabos/devices/workstation/conductivity_station/conductivity_station.json deleted file mode 100644 index 6d4194330..000000000 --- a/unilabos/devices/workstation/conductivity_station/conductivity_station.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "nodes": [ - { - "id": "CONDUCTIVITY_STATION", - "name": "电导率自动化测试工站", - "parent": null, - "type": "device", - "class": "conductivity_station", - "position": { - "x": 720.0, - "y": 200.0, - "z": 0 - }, - "config": { - "ip": "127.0.0.1", - "port": 19091, - "connect_timeout": 5.0, - "response_timeout": 10.0, - "max_message_bytes": 4194304, - "encoding": "utf-8", - "frame_delimiter": "\\r\\n", - "station_action_names": {} - }, - "data": {}, - "children": [] - } - ], - "links": [] -} From d038b0ebfa9168aed80f6054d39c18bd133cd0ea Mon Sep 17 00:00:00 2001 From: IkukiKanade <734588986@qq.com> Date: Tue, 18 Aug 2026 17:20:28 +0800 Subject: [PATCH 3/4] refactor: remove conductivity driver from core --- tests/devices/test_conductivity_station.py | 188 -------- .../test_conductivity_station_registry.py | 78 --- .../conductivity_station/__init__.py | 5 - .../conductivity_station.py | 221 --------- .../conductivity_station/mock_server.py | 380 --------------- .../conductivity_station/mock_unilab.py | 448 ------------------ 6 files changed, 1320 deletions(-) delete mode 100644 tests/devices/test_conductivity_station.py delete mode 100644 tests/devices/test_conductivity_station_registry.py delete mode 100644 unilabos/devices/workstation/conductivity_station/__init__.py delete mode 100644 unilabos/devices/workstation/conductivity_station/conductivity_station.py delete mode 100644 unilabos/devices/workstation/conductivity_station/mock_server.py delete mode 100644 unilabos/devices/workstation/conductivity_station/mock_unilab.py diff --git a/tests/devices/test_conductivity_station.py b/tests/devices/test_conductivity_station.py deleted file mode 100644 index 097d198bf..000000000 --- a/tests/devices/test_conductivity_station.py +++ /dev/null @@ -1,188 +0,0 @@ -"""电导工站 TCP 客户端与模拟服务端集成测试。""" - -from __future__ import annotations - -import json -import math -import socketserver -import threading -import time - -import pytest - -from unilabos.devices.workstation.conductivity_station.conductivity_station import ( - ConductivityStation, - ConductivityStationProtocolError, - ConductivityStationTransportError, -) -from unilabos.devices.workstation.conductivity_station.mock_server import ( - MockConductivityServer, - MockConductivityState, -) - - -@pytest.fixture() -def station() -> ConductivityStation: - state = MockConductivityState(step_interval=0.01) - server = MockConductivityServer(("127.0.0.1", 0), state=state) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - client = ConductivityStation( - ip="127.0.0.1", - port=server.server_address[1], - response_timeout=1.0, - ) - try: - yield client - finally: - client.close() - server.shutdown() - server.server_close() - thread.join(timeout=2) - - -def test_status_material_start_and_result(station: ConductivityStation) -> None: - assert station.station_status()["data"]["robot_arm"] == 1 - assert station.material_status()["data"]["bottle"][:3] == [1, 1, 1] - - started = station.start_batch() - assert started["result"] == 0 - batch_id = started["data"]["batch_id"] - - deadline = time.monotonic() + 2 - while time.monotonic() < deadline: - status = station.batch_status()["data"] - if not status["running"]: - break - time.sleep(0.02) - else: - pytest.fail("模拟批次未在预期时间内完成") - assert status["batch"]["current_test"] == 3 - assert status["batch"]["current_test_step"] == 13 - - result = station.batch_result(batch_id) - tests = result["data"]["batch"]["tests"] - assert [item["state"] for item in tests] == ["finished"] * 3 - assert tests[0]["bottle_code"].startswith("LOT-") - assert all(item["test_time"] for item in tests) - assert tests[0]["R"] > 0 - assert tests[0]["ion_conductivity"] > 0 - assert tests[0]["area"] == pytest.approx(math.pi * 0.5**2) - thickness_cm = (tests[0]["h2"] - tests[0]["h1"]) / 10.0 - expected_ms_cm = thickness_cm / (tests[0]["R"] * tests[0]["area"]) * 1000 - assert tests[0]["ion_conductivity"] == pytest.approx(expected_ms_cm, rel=1e-5) - - -def test_manual_validation_and_stop_after_current_sample( - station: ConductivityStation, -) -> None: - with pytest.raises(ValueError, match="1 到 13"): - station.manual_run(14) - - batch_id = station.start_batch()["data"]["batch_id"] - assert station.manual_run(6)["result"] == 0 - assert station.stop_current_batch()["result"] == 0 - - deadline = time.monotonic() + 1 - while time.monotonic() < deadline: - status = station.batch_status()["data"] - if not status["running"]: - break - time.sleep(0.02) - assert status["batch"]["state"] == "stopped" - tests = station.batch_result(batch_id)["data"]["batch"]["tests"] - assert tests[0]["state"] == "finished" - assert tests[1]["state"] == "not_started" - - -def test_unknown_batch_and_clear(station: ConductivityStation) -> None: - assert station.batch_result("missing")["result"] == 9 - assert station.clear_current_batch()["result"] == 0 - status = station.batch_status()["data"] - assert status == {"running": False, "batch": None} - - -def _scripted_station(plans: list[object]) -> tuple[ConductivityStation, socketserver.TCPServer, threading.Thread]: - """启动一个按连接顺序返回分片、断线或指定响应的协议测试服务。""" - - class Handler(socketserver.BaseRequestHandler): - def handle(self) -> None: - received = bytearray() - while b"\r\n" not in received: - chunk = self.request.recv(4096) - if not chunk: - return - received.extend(chunk) - request = json.loads(bytes(received).split(b"\r\n", 1)[0]) - with self.server.plan_lock: # type: ignore[attr-defined] - plan = self.server.plans.pop(0) # type: ignore[attr-defined] - if plan is None: - return - if callable(plan): - plan = plan(request) - chunks = plan if isinstance(plan, list) else [plan] - for chunk in chunks: - self.request.sendall(chunk) - - server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler) - server.plans = list(plans) # type: ignore[attr-defined] - server.plan_lock = threading.Lock() # type: ignore[attr-defined] - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - client = ConductivityStation( - ip="127.0.0.1", port=server.server_address[1], response_timeout=0.5 - ) - return client, server, thread - - -def _close_scripted_station( - client: ConductivityStation, - server: socketserver.TCPServer, - thread: threading.Thread, -) -> None: - client.close() - server.shutdown() - server.server_close() - thread.join(timeout=2) - - -def test_tcp_fragmentation_and_read_retry() -> None: - def fragmented(request: dict) -> list[bytes]: - payload = json.dumps( - {"request_id": request["request_id"], "result": 0, "data": {"running": False, "batch": None}} - ).encode() - return [payload[:8], payload[8:] + b"\r\n"] - - client, server, thread = _scripted_station([None, fragmented]) - try: - # 查询动作首次断线后可安全重连，且能拼接分片响应。 - assert client.batch_status()["data"]["running"] is False - assert server.plans == [] # type: ignore[attr-defined] - finally: - _close_scripted_station(client, server, thread) - -@pytest.mark.parametrize( - "response,exception", - [ - (b'{"request_id":999,"result":0}\r\n', ConductivityStationProtocolError), - (b'{not-json}\r\n', ConductivityStationProtocolError), - ], -) -def test_invalid_responses_are_rejected(response: bytes, exception: type[Exception]) -> None: - client, server, thread = _scripted_station([response]) - try: - with pytest.raises(exception): - client.start_batch() - finally: - _close_scripted_station(client, server, thread) - - -def test_side_effect_action_is_not_retried_after_disconnect() -> None: - client, server, thread = _scripted_station([None, b'{"request_id":1,"result":0}\r\n']) - try: - with pytest.raises(ConductivityStationTransportError): - client.start_batch() - # 第二个计划仍在，证明有副作用动作未被自动重发。 - assert len(server.plans) == 1 # type: ignore[attr-defined] - finally: - _close_scripted_station(client, server, thread) diff --git a/tests/devices/test_conductivity_station_registry.py b/tests/devices/test_conductivity_station_registry.py deleted file mode 100644 index ca6cb8cb7..000000000 --- a/tests/devices/test_conductivity_station_registry.py +++ /dev/null @@ -1,78 +0,0 @@ -from __future__ import annotations - -import ast -from pathlib import Path - - -ROOT = Path(__file__).resolve().parents[2] -DEVICE_DIR = ( - ROOT - / "unilabos" - / "devices" - / "workstation" - / "conductivity_station" -) - - -def test_conductivity_station_actions_are_discoverable_by_registry() -> None: - tree = ast.parse( - (DEVICE_DIR / "conductivity_station.py").read_text(encoding="utf-8") - ) - device_class = next( - node - for node in tree.body - if isinstance(node, ast.ClassDef) and node.name == "ConductivityStation" - ) - device_decorator = next( - decorator - for decorator in device_class.decorator_list - if isinstance(decorator, ast.Call) - and isinstance(decorator.func, ast.Name) - and decorator.func.id == "device" - ) - device_id = next( - keyword.value.value - for keyword in device_decorator.keywords - if keyword.arg == "id" and isinstance(keyword.value, ast.Constant) - ) - actions = { - method.name - for method in device_class.body - if isinstance(method, ast.FunctionDef) - and any( - (isinstance(decorator, ast.Name) and decorator.id == "action") - or ( - isinstance(decorator, ast.Call) - and isinstance(decorator.func, ast.Name) - and decorator.func.id == "action" - ) - for decorator in method.decorator_list - ) - } - init_method = next( - method - for method in device_class.body - if isinstance(method, ast.FunctionDef) and method.name == "__init__" - ) - - assert device_id == "conductivity_station" - assert actions == { - "station_status", - "material_status", - "batch_status", - "batch_result", - "start_batch", - "stop_current_batch", - "manual_run", - "clear_current_batch", - } - assert {arg.arg for arg in init_method.args.args if arg.arg != "self"} == { - "ip", - "port", - "connect_timeout", - "response_timeout", - "max_message_bytes", - "encoding", - "frame_delimiter", - "station_action_names", - } diff --git a/unilabos/devices/workstation/conductivity_station/__init__.py b/unilabos/devices/workstation/conductivity_station/__init__.py deleted file mode 100644 index ef00f5920..000000000 --- a/unilabos/devices/workstation/conductivity_station/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""电导率自动化工站 TCP 设备。""" - -from .conductivity_station import ConductivityStation - -__all__ = ["ConductivityStation"] diff --git a/unilabos/devices/workstation/conductivity_station/conductivity_station.py b/unilabos/devices/workstation/conductivity_station/conductivity_station.py deleted file mode 100644 index 47acae374..000000000 --- a/unilabos/devices/workstation/conductivity_station/conductivity_station.py +++ /dev/null @@ -1,221 +0,0 @@ -"""电导率自动化工站 TCP 客户端及 UniLab 动作。 - -合作方工站作为 TCP 服务端，UniLab 作为客户端。报文为 UTF-8 JSON，使用 -CRLF 分帧。查询类动作允许在连接中断后重连重试一次；启动、停止和手动运行 -属于有副作用的命令，发送结果不明确时不会自动重发，避免重复执行。 -""" - -from __future__ import annotations - -import json -import socket -import threading -from itertools import count -from typing import Any - -from unilabos.registry.decorators import action, device - - -class ConductivityStationTransportError(RuntimeError): - """TCP 连接或报文传输失败。""" - - -class ConductivityStationProtocolError(RuntimeError): - """服务端响应不符合电导工站协议。""" - - -@device( - id="conductivity_station", - category=["workstation", "conductivity"], - displayname="电导率自动化测试工站", - description="通过 TCP JSON/CRLF 协议控制电导率自动化测试工站", - version="1.0.0", -) -class ConductivityStation: - """电导工站长连接客户端。""" - - def __init__( - self, - ip: str = "127.0.0.1", - port: int = 19091, - connect_timeout: float = 5.0, - response_timeout: float = 10.0, - max_message_bytes: int = 4194304, - encoding: str = "utf-8", - frame_delimiter: str = "\\r\\n", - station_action_names: dict[str, str] | None = None, - **_: Any, - ) -> None: - self.ip = str(ip) - self.port = int(port) - self.connect_timeout = float(connect_timeout) - self.response_timeout = float(response_timeout) - self.max_message_bytes = int(max_message_bytes) - self.encoding = str(encoding).strip() or "utf-8" - delimiter_text = ( - str(frame_delimiter).replace("\\r", "\r").replace("\\n", "\n") - ) - self.frame_delimiter = delimiter_text.encode(self.encoding) - if not self.frame_delimiter: - raise ValueError("frame_delimiter 不能为空") - self.station_action_names = { - str(key): str(value) - for key, value in (station_action_names or {}).items() - if str(key) and str(value) - } - self.status = "idle" - self._sock: socket.socket | None = None - self._recv_buffer = bytearray() - self._request_ids = count(1) - self._lock = threading.RLock() - - def close(self) -> None: - """关闭当前 TCP 连接。""" - with self._lock: - self._disconnect() - - def _disconnect(self) -> None: - sock, self._sock = self._sock, None - self._recv_buffer.clear() - if sock is not None: - try: - sock.shutdown(socket.SHUT_RDWR) - except OSError: - pass - try: - sock.close() - except OSError: - pass - - def _connect(self) -> socket.socket: - if self._sock is None: - try: - sock = socket.create_connection( - (self.ip, self.port), timeout=self.connect_timeout - ) - sock.settimeout(self.response_timeout) - self._sock = sock - except OSError as exc: - self._disconnect() - raise ConductivityStationTransportError( - f"无法连接电导工站 {self.ip}:{self.port}: {exc}" - ) from exc - return self._sock - - def _read_frame(self, sock: socket.socket) -> bytes: - delimiter = self.frame_delimiter - while True: - marker = self._recv_buffer.find(delimiter) - if marker >= 0: - frame = bytes(self._recv_buffer[:marker]) - del self._recv_buffer[: marker + len(delimiter)] - return frame - if len(self._recv_buffer) > self.max_message_bytes: - raise ConductivityStationProtocolError( - f"响应超过最大长度 {self.max_message_bytes} 字节" - ) - try: - chunk = sock.recv(65536) - except (OSError, socket.timeout) as exc: - raise ConductivityStationTransportError( - f"等待电导工站响应失败: {exc}" - ) from exc - if not chunk: - raise ConductivityStationTransportError("电导工站在响应前关闭连接") - self._recv_buffer.extend(chunk) - - def _exchange(self, request: dict[str, Any]) -> dict[str, Any]: - sock = self._connect() - encoded = json.dumps( - request, ensure_ascii=False, separators=(",", ":") - ).encode(self.encoding) + self.frame_delimiter - if len(encoded) > self.max_message_bytes: - raise ConductivityStationProtocolError( - f"请求超过最大长度 {self.max_message_bytes} 字节" - ) - try: - sock.sendall(encoded) - frame = self._read_frame(sock) - except Exception: - self._disconnect() - raise - try: - response = json.loads(frame.decode(self.encoding)) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - self._disconnect() - raise ConductivityStationProtocolError( - f"响应不是有效 {self.encoding} JSON: {exc}" - ) from exc - if not isinstance(response, dict): - raise ConductivityStationProtocolError("响应必须是 JSON 对象") - if response.get("request_id") != request["request_id"]: - raise ConductivityStationProtocolError( - "响应 request_id 不匹配: " - f"期望 {request['request_id']}，实际 {response.get('request_id')}" - ) - if "result" not in response: - raise ConductivityStationProtocolError("响应缺少 result 字段") - return response - - def _request( - self, - action_name: str, - param: dict[str, Any] | None = None, - *, - retry_read: bool = False, - ) -> dict[str, Any]: - with self._lock: - request: dict[str, Any] = { - "request_id": next(self._request_ids), - "action": self.station_action_names.get(action_name, action_name), - } - if param is not None: - request["param"] = param - attempts = 2 if retry_read else 1 - for attempt in range(attempts): - try: - return self._exchange(request) - except ConductivityStationTransportError: - if attempt + 1 >= attempts: - raise - self._disconnect() - raise AssertionError("unreachable") - - @action(always_free=True, description="查询电导工站各机构在线状态") - def station_status(self) -> dict[str, Any]: - return self._request("station_status", retry_read=True) - - @action(always_free=True, description="查询料瓶、模具和漏斗占位情况") - def material_status(self) -> dict[str, Any]: - return self._request("material_status", retry_read=True) - - @action(always_free=True, description="查询当前电导测试批次和执行进度") - def batch_status(self) -> dict[str, Any]: - return self._request("batch_status", retry_read=True) - - @action(always_free=True, description="按批次号查询样品测试结果") - def batch_result(self, batch_id: str) -> dict[str, Any]: - if not str(batch_id).strip(): - raise ValueError("batch_id 不能为空") - return self._request( - "batch_result", {"batch_id": str(batch_id).strip()}, retry_read=True - ) - - @action(description="上料完成后启动整批自动测试") - def start_batch(self) -> dict[str, Any]: - return self._request("start_batch") - - @action(description="当前样品完成后停止整批任务") - def stop_current_batch(self) -> dict[str, Any]: - return self._request("stop_current_batch") - - @action(description="切换手动模式并单独执行指定步骤") - def manual_run(self, step: int) -> dict[str, Any]: - step_value = int(step) - if not 1 <= step_value <= 13: - raise ValueError("step 必须在 1 到 13 之间") - return self._request("manual_run", {"step": step_value}) - - @action(description="清理非运行状态下的当前批次（动作名待合作方最终确认）") - def clear_current_batch(self) -> dict[str, Any]: - return self._request("clear_current_batch") diff --git a/unilabos/devices/workstation/conductivity_station/mock_server.py b/unilabos/devices/workstation/conductivity_station/mock_server.py deleted file mode 100644 index 6ff359890..000000000 --- a/unilabos/devices/workstation/conductivity_station/mock_server.py +++ /dev/null @@ -1,380 +0,0 @@ -"""电导工站 TCP 协议模拟服务端。""" - -from __future__ import annotations - -import argparse -import copy -import json -import math -import socketserver -import threading -import time -from datetime import datetime -from typing import Any - - -STEP_NAMES = { - 1: "模具拆解", - 2: "模具测厚", - 3: "模具转移", - 4: "漏斗转移", - 5: "烧结料瓶扫码与转移", - 6: "加粉", - 7: "烧结料瓶暂存", - 8: "模具组装", - 9: "模具转移", - 10: "EIS测试", - 11: "模具测厚", - 12: "电子电导率测试", - 13: "测试物料转移至托盘", -} - -# 合作方高保真模型确认使用直径 1 cm 圆片：A = π × (0.5 cm)²。 -DISC_AREA_CM2 = math.pi * 0.5**2 - - -class MockConductivityState: - """线程安全的批次模拟状态。""" - - def __init__( - self, - *, - step_interval: float = 0.1, - failure_sample: int | None = None, - failure_step: int = 10, - ) -> None: - self.step_interval = max(0.01, float(step_interval)) - self.failure_sample = failure_sample - self.failure_step = int(failure_step) - self.lock = threading.RLock() - self.started_monotonic: float | None = None - self.stop_requested = False - self.stop_after_sample: int | None = None - self.batch: dict[str, Any] | None = None - self.history: dict[str, dict[str, Any]] = {} - self.materials = { - "bottle": [1, 1, 1, 0, 0, 0, 0, 0, 0, 0], - "mold": [0, 0, 0, 1, 1, 1, 0, 0, 0, 0], - "funnel": [0, 0, 0, 0, 0, 0, 1, 1, 1, 0], - } - - @staticmethod - def _positions(values: list[int]) -> list[int]: - return [index for index, occupied in enumerate(values, start=1) if occupied] - - def _new_test(self, index: int, bottle: int, mold: int, funnel: int) -> dict[str, Any]: - return { - "state": "not_started", - "test_time": "", - "bottle": bottle, - "mold": mold, - "funnel": funnel, - "step": 0, - "bottle_code": "", - "recipe": "", - "formula": "", - "temperature": 0.0, - "due_pressure": 3, - "pressure_time": 10, - "R": 0.0, - "h1": 0.0, - "h2": 0.0, - "area": DISC_AREA_CM2, - "ion_conductivity": 0.0, - "elec_conductivity_tested": False, - "elec_R": 0.0, - "elec_conductivity": 0.0, - "_sample_index": index, - } - - def _update(self) -> None: - if not self.batch or self.started_monotonic is None or not self.batch["running"]: - return - tests = self.batch["tests"] - elapsed_steps = int((time.monotonic() - self.started_monotonic) / self.step_interval) - target_completed = elapsed_steps // 13 - active_step = elapsed_steps % 13 + 1 - - if self.stop_after_sample is not None: - target_completed = min(target_completed, self.stop_after_sample) - - for index, test in enumerate(tests, start=1): - if test["state"] in {"finished", "failed"}: - continue - failure_reached = ( - index == self.failure_sample - and elapsed_steps >= (index - 1) * 13 + self.failure_step - 1 - ) - if failure_reached: - if not test["test_time"]: - test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - test["state"] = "failed" - test["step"] = self.failure_step - test["error_code"] = "MOCK_STEP_FAILED" - test["error_message"] = f"模拟失败：{STEP_NAMES[self.failure_step]}" - self.batch["running"] = False - self.batch["state"] = "failed" - break - if index <= target_completed: - self._finish_test(test) - elif index == target_completed + 1 and not ( - self.stop_after_sample is not None - and index > self.stop_after_sample - ): - if not test["test_time"]: - test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - test["state"] = "in_progress" - test["step"] = active_step - break - - finished = sum(item["state"] == "finished" for item in tests) - failed = sum(item["state"] == "failed" for item in tests) - self.batch["finished_count"] = finished - self.batch["failed_count"] = failed - active = next( - (item for item in tests if item["state"] == "in_progress"), None - ) - if active: - self.batch["current_test"] = active["_sample_index"] - self.batch["current_test_step"] = active["step"] - if ( - self.stop_after_sample is not None - and finished >= self.stop_after_sample - ): - self.batch["running"] = False - self.batch["state"] = "stopped" - elif finished + failed == len(tests): - self.batch["running"] = False - self.batch["state"] = "completed" if failed == 0 else "failed" - self.batch["current_test"] = len(tests) - self.batch["current_test_step"] = tests[-1]["step"] if tests else 0 - self.history[self.batch["batch_id"]] = copy.deepcopy(self.batch) - - @staticmethod - def _finish_test(test: dict[str, Any]) -> None: - index = int(test["_sample_index"]) - if not test["test_time"]: - test["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - resistance_ohm = round(80.0 + index * 4.25, 3) - h1_mm = 8.0 - h2_mm = round(8.8 + index * 0.02, 3) - thickness_cm = (h2_mm - h1_mm) / 10.0 - ionic_conductivity_ms_cm = ( - thickness_cm / (resistance_ohm * DISC_AREA_CM2) * 1000.0 - ) - test.update( - { - "state": "finished", - "step": 13, - "bottle_code": f"LOT-260812-{index:03d}-CRU-S-{index:03d}-260812", - "recipe": "R-202608-001", - "formula": "Li6PS5Cl", - "temperature": round(26.5 + index * 0.1, 2), - "R": resistance_ohm, - "h1": h1_mm, - "h2": h2_mm, - "area": DISC_AREA_CM2, - "ion_conductivity": round(ionic_conductivity_ms_cm, 6), - } - ) - - @staticmethod - def _public(value: Any) -> Any: - if isinstance(value, dict): - return { - key: MockConductivityState._public(item) - for key, item in value.items() - if not key.startswith("_") - } - if isinstance(value, list): - return [MockConductivityState._public(item) for item in value] - return value - - def handle(self, request: dict[str, Any]) -> dict[str, Any]: - request_id = request.get("request_id") - action = request.get("action") - param = request.get("param") or {} - with self.lock: - self._update() - if action == "station_status": - return self._response( - request_id, - data={ - "robot_arm": 1, - "scanner": 1, - "lid_open_close_mechanism": 1, - "powder_adding_mechanism": 1, - "tablet_pressing_mechanism": 1, - "electrochemical_workstation": 1, - "stack_rack": 1, - }, - ) - if action == "material_status": - return self._response(request_id, data=copy.deepcopy(self.materials)) - if action == "batch_status": - if not self.batch: - return self._response(request_id, data={"running": False, "batch": None}) - summary_keys = ( - "batch_id", - "test_round", - "bottle", - "mold", - "funnel", - "current_test", - "current_test_step", - "finished_count", - "failed_count", - "state", - ) - summary = {key: self.batch.get(key) for key in summary_keys} - return self._response( - request_id, - data={"running": self.batch["running"], "batch": summary}, - ) - if action == "batch_result": - batch_id = str(param.get("batch_id") or "") - batch = self.history.get(batch_id) - if not batch: - return self._response(request_id, result=9) - return self._response( - request_id, - data={ - "batch": { - "batch_id": batch_id, - "test_round": batch["test_round"], - "state": batch["state"], - "tests": self._public(copy.deepcopy(batch["tests"])), - } - }, - ) - if action == "start_batch": - if self.batch and self.batch["running"]: - return self._response(request_id, result=4) - positions = [self._positions(self.materials[key]) for key in ("bottle", "mold", "funnel")] - if not positions[0] or len({len(items) for items in positions}) != 1: - return self._response(request_id, result=2) - batch_id = datetime.now().strftime("%Y%m%d%H%M%S%f")[:17] - tests = [ - self._new_test(index, bottle, mold, funnel) - for index, (bottle, mold, funnel) in enumerate(zip(*positions), start=1) - ] - self.batch = { - "batch_id": batch_id, - "test_round": len(tests), - "bottle": positions[0], - "mold": positions[1], - "funnel": positions[2], - "current_test": 1, - "current_test_step": 1, - "finished_count": 0, - "failed_count": 0, - "running": True, - "state": "running", - "tests": tests, - } - self.started_monotonic = time.monotonic() - self.stop_requested = False - self.stop_after_sample = None - self.history[batch_id] = copy.deepcopy(self.batch) - return self._response(request_id, data={"batch_id": batch_id}) - if action == "stop_current_batch": - if not self.batch or not self.batch["running"]: - return self._response(request_id, result=3) - self.stop_requested = True - self.stop_after_sample = int(self.batch.get("current_test") or 1) - self.batch["state"] = "stopping" - return self._response(request_id) - if action == "manual_run": - if not self.batch or not self.batch["running"]: - return self._response(request_id, result=3) - step = int(param.get("step") or 0) - if not 1 <= step <= 13: - return self._response(request_id, result=5) - active = next( - (item for item in self.batch["tests"] if item["state"] == "in_progress"), - self.batch["tests"][0], - ) - active["state"] = "in_progress" - active["step"] = step - if not active["test_time"]: - active["test_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - return self._response( - request_id, - data={ - "batch_id": self.batch["batch_id"], - "test_time": active["test_time"], - }, - ) - if action == "clear_current_batch": - if self.batch and self.batch["running"]: - return self._response(request_id, result=4) - self.batch = None - self.started_monotonic = None - self.stop_requested = False - self.stop_after_sample = None - return self._response(request_id) - return self._response(request_id, result=8) - - @staticmethod - def _response( - request_id: Any, result: int = 0, data: dict[str, Any] | None = None - ) -> dict[str, Any]: - response: dict[str, Any] = {"request_id": request_id, "result": result} - if data is not None: - response["data"] = data - return response - - -class _Handler(socketserver.StreamRequestHandler): - def handle(self) -> None: - state: MockConductivityState = self.server.state # type: ignore[attr-defined] - while True: - raw = self.rfile.readline() - if not raw: - return - try: - request = json.loads(raw.decode("utf-8").strip()) - if not isinstance(request, dict): - raise ValueError("请求必须是 JSON 对象") - response = state.handle(request) - except Exception as exc: # noqa: BLE001 - response = {"request_id": None, "result": 7, "message": str(exc)} - payload = json.dumps(response, ensure_ascii=False, separators=(",", ":")) - self.wfile.write((payload + "\r\n").encode("utf-8")) - self.wfile.flush() - - -class MockConductivityServer(socketserver.ThreadingTCPServer): - allow_reuse_address = True - daemon_threads = True - - def __init__( - self, - address: tuple[str, int] = ("127.0.0.1", 19091), - *, - state: MockConductivityState | None = None, - ) -> None: - self.state = state or MockConductivityState() - super().__init__(address, _Handler) - - -def main() -> None: - parser = argparse.ArgumentParser(description="电导工站 TCP 协议模拟服务端") - parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--port", type=int, default=19091) - parser.add_argument("--step-interval", type=float, default=0.5) - parser.add_argument("--failure-sample", type=int) - parser.add_argument("--failure-step", type=int, default=10) - args = parser.parse_args() - state = MockConductivityState( - step_interval=args.step_interval, - failure_sample=args.failure_sample, - failure_step=args.failure_step, - ) - with MockConductivityServer((args.host, args.port), state=state) as server: - print(f"模拟电导工站监听 {args.host}:{server.server_address[1]}", flush=True) - server.serve_forever() - - -if __name__ == "__main__": - main() diff --git a/unilabos/devices/workstation/conductivity_station/mock_unilab.py b/unilabos/devices/workstation/conductivity_station/mock_unilab.py deleted file mode 100644 index 98d82bef4..000000000 --- a/unilabos/devices/workstation/conductivity_station/mock_unilab.py +++ /dev/null @@ -1,448 +0,0 @@ -"""用于电导工站联调的最小 UniLab Job API 模拟网关。 - -该网关只用于没有编译 ``unilabos_msgs`` 的本地开发环境。平台仍使用正式的 -UniLab Job API 契约；网关把 Job 动作分发到 ``ConductivityStation`` 注册设备， -设备再通过真实 TCP/CRLF 协议访问模拟工站。 -""" - -from __future__ import annotations - -import argparse -import html -import inspect -import json -import threading -import uuid -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from typing import Any -from urllib.parse import urlparse - -from .conductivity_station import ConductivityStation -from .mock_server import MockConductivityServer, MockConductivityState - - -class MockUniLabJobState: - """线程安全的 Job 存储与动作分发器。""" - - def __init__( - self, - device: ConductivityStation, - device_id: str = "CONDUCTIVITY_STATION", - ) -> None: - self.device = device - self.device_id = device_id - self.jobs: dict[str, dict[str, Any]] = {} - self.lock = threading.RLock() - - @staticmethod - def _json_type(annotation: Any) -> str: - """将动作参数注解转换为页面与 API 使用的简化 JSON Schema 类型。""" - text = str(annotation).lower() - if annotation is int or text in {"int", ""}: - return "integer" - if annotation is float or text in {"float", ""}: - return "number" - if annotation is bool or text in {"bool", ""}: - return "boolean" - if "list" in text or "tuple" in text or "set" in text: - return "array" - if "dict" in text: - return "object" - return "string" - - def actions(self) -> dict[str, dict[str, Any]]: - """从实际设备类的 ``@action`` 元数据动态生成动作目录。""" - catalog: dict[str, dict[str, Any]] = {} - for name, method in inspect.getmembers(type(self.device), inspect.isfunction): - meta = getattr(method, "_action_registry_meta", None) - if not isinstance(meta, dict): - continue - signature = inspect.signature(method) - properties: dict[str, dict[str, Any]] = {} - required: list[str] = [] - parameters: list[dict[str, Any]] = [] - for parameter in signature.parameters.values(): - if parameter.name == "self": - continue - parameter_type = self._json_type(parameter.annotation) - is_required = parameter.default is inspect.Parameter.empty - item: dict[str, Any] = { - "name": parameter.name, - "type": parameter_type, - "required": is_required, - } - schema_item: dict[str, Any] = {"type": parameter_type} - if is_required: - required.append(parameter.name) - else: - item["default"] = parameter.default - schema_item["default"] = parameter.default - parameters.append(item) - properties[parameter.name] = schema_item - schema: dict[str, Any] = { - "type": "object", - "properties": properties, - "additionalProperties": False, - } - if required: - schema["required"] = required - catalog[name] = { - "name": name, - "description": str(meta.get("description") or ""), - "always_free": bool(meta.get("always_free")), - "parameters": parameters, - "schema": schema, - } - return dict(sorted(catalog.items())) - - def submit( - self, device_id: str, action_name: str, action_args: dict[str, Any] - ) -> dict[str, Any]: - job_id = str(uuid.uuid4()) - if device_id != self.device_id: - result = { - "jobId": job_id, - "status": 6, - "result": {"error": f"Device not found: {device_id}"}, - } - with self.lock: - self.jobs[job_id] = result - return result - method = getattr(self.device, action_name, None) - if action_name not in self.actions() or method is None: - result = { - "jobId": job_id, - "status": 6, - "result": {"error": f"Action not found: {action_name}"}, - } - with self.lock: - self.jobs[job_id] = result - return result - try: - return_value = method(**action_args) - result = { - "jobId": job_id, - "status": 4, - "result": {"return_value": return_value}, - } - except Exception as exc: # noqa: BLE001 - result = { - "jobId": job_id, - "status": 6, - "result": {"error": str(exc)}, - } - with self.lock: - self.jobs[job_id] = result - return result - - def status(self, job_id: str) -> dict[str, Any]: - with self.lock: - return self.jobs.get( - job_id, - { - "jobId": job_id, - "status": 6, - "result": {"error": "Job not found"}, - }, - ) - - -class _JobApiHandler(BaseHTTPRequestHandler): - server_version = "MockUniLab/1.0" - - def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - return - - @property - def state(self) -> MockUniLabJobState: - return self.server.state # type: ignore[attr-defined,no-any-return] - - def _json_body(self) -> dict[str, Any]: - length = int(self.headers.get("Content-Length") or 0) - raw = self.rfile.read(length) if length else b"{}" - value = json.loads(raw.decode("utf-8")) - if not isinstance(value, dict): - raise ValueError("请求体必须是 JSON 对象") - return value - - def _send(self, data: Any, *, code: int = 0, message: str = "success") -> None: - payload = json.dumps( - {"code": code, "data": data, "message": message}, - ensure_ascii=False, - separators=(",", ":"), - ).encode("utf-8") - self.send_response(200) - self.send_header("Content-Type", "application/json; charset=utf-8") - self.send_header("Content-Length", str(len(payload))) - self.end_headers() - self.wfile.write(payload) - - def _send_html(self, content: str) -> None: - payload = content.encode("utf-8") - self.send_response(200) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(payload))) - self.end_headers() - self.wfile.write(payload) - - def _status_page(self) -> str: - action_rows = [] - for action in self.state.actions().values(): - parameters = ", ".join( - f"{item['name']}: {item['type']}" - for item in action["parameters"] - ) or "无" - mode = "只读/免排队" if action["always_free"] else "控制动作" - action_rows.append( - "" - f"{html.escape(action['name'])}" - f"{html.escape(parameters)}" - f"{html.escape(action['description'])}" - f"{mode}" - "" - ) - rows = "".join(action_rows) - device_id = html.escape(self.state.device_id) - return f""" - - - - - Mock UniLab 主机状态 - - - -
-

UniLab 主机状态 · Mock

-

电导工站本地接口模拟与动作注册展示

-
-
模拟环境：此页面不是完整 UniLab/ROS 运行时，仅用于验证驱动动作发现和 Job API 联调。
-
-
-
{device_id}Onlinelocal-mock
-

设备类：conductivity_station　动作数量：{len(action_rows)}

-
-
-

已发现的注册动作

- - - {rows} -
动作参数说明模式
-
-
- 动作 API： - /api/v1/devices/{device_id}/actions -
-
- -""" - - def do_POST(self) -> None: # noqa: N802 - if urlparse(self.path).path != "/api/v1/job/add": - self._send({}, code=1, message="not found") - return - try: - body = self._json_body() - device_id = str(body.get("device_id") or "") - action_name = str(body.get("action") or "") - action_args = body.get("action_args") or {} - if not isinstance(action_args, dict): - raise ValueError("action_args 必须是对象") - job = self.state.submit(device_id, action_name, action_args) - self._send({"jobId": job["jobId"], "status": 1}) - except Exception as exc: # noqa: BLE001 - self._send({}, code=1, message=str(exc)) - - def do_GET(self) -> None: # noqa: N802 - path = urlparse(self.path).path - if path in {"/", "/status", "/registry-editor"}: - self._send_html(self._status_page()) - return - if path == "/api/v1/online-devices": - self._send( - { - "online_devices": { - self.state.device_id: { - "device_key": f"/devices/{self.state.device_id}", - "namespace": "/devices", - "machine_name": "local-mock", - } - }, - "total_count": 1, - } - ) - return - action_prefix = f"/api/v1/devices/{self.state.device_id}/actions" - if path == action_prefix: - actions = self.state.actions() - self._send( - { - "device_id": self.state.device_id, - "actions": actions, - "total_count": len(actions), - "environment": "mock", - } - ) - return - schema_marker = action_prefix + "/" - schema_suffix = "/schema" - if path.startswith(schema_marker) and path.endswith(schema_suffix): - action_name = path[len(schema_marker) : -len(schema_suffix)] - action = self.state.actions().get(action_name) - if action is None: - self._send({}, code=1, message=f"Action not found: {action_name}") - else: - self._send( - { - "device_id": self.state.device_id, - "action": action_name, - "schema": action["schema"], - } - ) - return - if path == "/api/v1/actions": - actions = self.state.actions() - self._send( - { - "devices": {self.state.device_id: actions}, - "total_count": len(actions), - "environment": "mock", - } - ) - return - prefix, suffix = "/api/v1/job/", "/status" - if path.startswith(prefix) and path.endswith(suffix): - job_id = path[len(prefix) : -len(suffix)] - self._send(self.state.status(job_id)) - return - self._send({}, code=1, message="not found") - - -class MockUniLabJobServer(ThreadingHTTPServer): - daemon_threads = True - allow_reuse_address = True - - def __init__( - self, - address: tuple[str, int], - *, - state: MockUniLabJobState, - ) -> None: - self.state = state - super().__init__(address, _JobApiHandler) - - -class ConductivityIntegrationMock: - """同时管理模拟工站和模拟 UniLab Job API，供测试及演示使用。""" - - def __init__( - self, - host: str = "127.0.0.1", - api_port: int = 18002, - station_port: int = 19091, - step_interval: float = 0.1, - failure_sample: int | None = None, - failure_step: int = 10, - ) -> None: - station_state = MockConductivityState( - step_interval=step_interval, - failure_sample=failure_sample, - failure_step=failure_step, - ) - self.station_server = MockConductivityServer( - (host, station_port), state=station_state - ) - actual_station_port = int(self.station_server.server_address[1]) - self.device = ConductivityStation(ip=host, port=actual_station_port) - self.job_state = MockUniLabJobState(self.device) - self.api_server = MockUniLabJobServer((host, api_port), state=self.job_state) - self._threads: list[threading.Thread] = [] - - @property - def api_port(self) -> int: - return int(self.api_server.server_address[1]) - - @property - def station_port(self) -> int: - return int(self.station_server.server_address[1]) - - def start(self) -> None: - for server, name in ( - (self.station_server, "mock-conductivity-station"), - (self.api_server, "mock-unilab-job-api"), - ): - thread = threading.Thread( - target=server.serve_forever, name=name, daemon=True - ) - thread.start() - self._threads.append(thread) - - def close(self) -> None: - self.device.close() - for server in (self.api_server, self.station_server): - server.shutdown() - server.server_close() - for thread in self._threads: - thread.join(timeout=2) - self._threads.clear() - - def __enter__(self) -> "ConductivityIntegrationMock": - self.start() - return self - - def __exit__(self, *_: Any) -> None: - self.close() - - -def main() -> None: - parser = argparse.ArgumentParser(description="电导工站 UniLab 全链路模拟器") - parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--api-port", type=int, default=18002) - parser.add_argument("--station-port", type=int, default=19091) - parser.add_argument("--step-interval", type=float, default=0.5) - parser.add_argument("--failure-sample", type=int) - parser.add_argument("--failure-step", type=int, default=10) - args = parser.parse_args() - mock = ConductivityIntegrationMock( - host=args.host, - api_port=args.api_port, - station_port=args.station_port, - step_interval=args.step_interval, - failure_sample=args.failure_sample, - failure_step=args.failure_step, - ) - mock.start() - print( - f"模拟 UniLab Job API: http://{args.host}:{mock.api_port}/api/v1\n" - f"模拟电导工站 TCP: {args.host}:{mock.station_port}", - flush=True, - ) - try: - threading.Event().wait() - except KeyboardInterrupt: - pass - finally: - mock.close() - - -if __name__ == "__main__": - main() From 8f96e09880429074db59bc23062b38c3164750eb Mon Sep 17 00:00:00 2001 From: IkukiKanade <734588986@qq.com> Date: Tue, 18 Aug 2026 17:20:28 +0800 Subject: [PATCH 4/4] fix: improve WSS and host action visibility --- tests/test_host_business_actions.py | 62 ++++++++++++++++++++++++++ tests/test_ws_client_ssl.py | 19 ++++++++ unilabos/app/ssl_utils.py | 12 +++++ unilabos/app/web/templates/status.html | 28 ++++++++++++ unilabos/app/web/utils/host_utils.py | 24 +++++++++- unilabos/app/ws_client.py | 9 ++-- unilabos/utils/requirements.txt | 3 +- 7 files changed, 151 insertions(+), 6 deletions(-) create mode 100644 tests/test_host_business_actions.py create mode 100644 tests/test_ws_client_ssl.py create mode 100644 unilabos/app/ssl_utils.py diff --git a/tests/test_host_business_actions.py b/tests/test_host_business_actions.py new file mode 100644 index 000000000..1c02dcb19 --- /dev/null +++ b/tests/test_host_business_actions.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import ModuleType, SimpleNamespace + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_host_status_exposes_json_business_actions(monkeypatch) -> None: + fake_host = SimpleNamespace( + devices_names={}, + _online_devices=set(), + device_machine_names={}, + _subscribed_topics=set(), + _action_clients={}, + _action_value_mappings={ + "CONDUCTIVITY_STATION": { + "station_status": { + "type": "UniLabJsonCommand", + "schema": {"description": "查询工站状态"}, + }, + "auto-close": {"type": "UniLabJsonCommand", "schema": {}}, + "legacy_action": {"type": "EmptyIn", "schema": {}}, + } + }, + device_status={}, + device_status_timestamps={}, + ) + + config_module = ModuleType("unilabos.config.config") + config_module.BasicConfig = SimpleNamespace(is_host_mode=True) + host_node_module = ModuleType("unilabos.ros.nodes.presets.host_node") + host_node_module.HostNode = SimpleNamespace(get_instance=lambda _timeout: fake_host) + action_utils_module = ModuleType("unilabos.app.web.utils.action_utils") + action_utils_module.get_action_info = lambda client, full_name: { + "client": client, + "full_name": full_name, + } + + monkeypatch.setitem(sys.modules, "unilabos.config.config", config_module) + monkeypatch.setitem(sys.modules, "unilabos.ros.nodes.presets.host_node", host_node_module) + monkeypatch.setitem(sys.modules, "unilabos.app.web.utils.action_utils", action_utils_module) + + module_path = ROOT / "unilabos" / "app" / "web" / "utils" / "host_utils.py" + spec = importlib.util.spec_from_file_location("test_host_utils", module_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + info = module.get_host_node_info() + + assert info["business_actions"] == { + "CONDUCTIVITY_STATION/station_status": { + "device_id": "CONDUCTIVITY_STATION", + "action_name": "station_status", + "action_type": "UniLabJsonCommand", + "description": "查询工站状态", + } + } diff --git a/tests/test_ws_client_ssl.py b/tests/test_ws_client_ssl.py new file mode 100644 index 000000000..a506de192 --- /dev/null +++ b/tests/test_ws_client_ssl.py @@ -0,0 +1,19 @@ +from unittest.mock import patch + +from unilabos.app.ssl_utils import create_wss_ssl_context + + +def test_wss_ssl_context_uses_certifi_ca_bundle() -> None: + expected_context = object() + + with ( + patch("unilabos.app.ssl_utils.certifi.where", return_value="certifi-ca.pem"), + patch( + "unilabos.app.ssl_utils.ssl.create_default_context", + return_value=expected_context, + ) as create_default_context, + ): + actual_context = create_wss_ssl_context() + + assert actual_context is expected_context + create_default_context.assert_called_once_with(cafile="certifi-ca.pem") diff --git a/unilabos/app/ssl_utils.py b/unilabos/app/ssl_utils.py new file mode 100644 index 000000000..0a610ff93 --- /dev/null +++ b/unilabos/app/ssl_utils.py @@ -0,0 +1,12 @@ +"""WebSocket TLS 工具。""" + +from __future__ import annotations + +import ssl + +import certifi + + +def create_wss_ssl_context() -> ssl.SSLContext: + """使用 certifi CA 构建 WSS 上下文，避免 Windows 证书库异常。""" + return ssl.create_default_context(cafile=certifi.where()) diff --git a/unilabos/app/web/templates/status.html b/unilabos/app/web/templates/status.html index a0d3dfbc9..4414d53ac 100644 --- a/unilabos/app/web/templates/status.html +++ b/unilabos/app/web/templates/status.html @@ -201,6 +201,34 @@

已接纳动作:

+ +
+

+ 业务动作 + {{ host_node_info.business_actions|length }} +

+ + + + + + + + {% for action_key, action_info in host_node_info.business_actions.items() %} + + + + + + + {% else %} + + + + {% endfor %} +
设备 ID动作名类型说明
{{ action_info.device_id }}{{ action_info.action_name }}{{ action_info.action_type }}{{ action_info.description }}
没有发现已注册的业务动作
+
+

diff --git a/unilabos/app/web/utils/host_utils.py b/unilabos/app/web/utils/host_utils.py index 1400893b7..765fc8050 100644 --- a/unilabos/app/web/utils/host_utils.py +++ b/unilabos/app/web/utils/host_utils.py @@ -21,7 +21,13 @@ def get_host_node_info() -> Dict[str, Any]: Returns: Dict: 包含主机节点信息的字典 """ - host_info = {"available": False, "devices": {}, "subscribed_topics": [], "action_clients": {}} + host_info = { + "available": False, + "devices": {}, + "subscribed_topics": [], + "action_clients": {}, + "business_actions": {}, + } if not BasicConfig.is_host_mode: return host_info # 尝试获取HostNode实例，设置超时为0秒 @@ -44,6 +50,22 @@ def get_host_node_info() -> Dict[str, Any]: for action_id, client in host_node._action_clients.items(): host_info["action_clients"][action_id] = get_action_info(client, full_name=action_id) + # @action 生成的 UniLabJsonCommand 复用底层 command ROS action，不会出现在 + # _action_clients 中；单独展示业务映射，避免主机页误认为动作未注册。 + for device_id, mappings in host_node._action_value_mappings.items(): + for action_name, mapping in (mappings or {}).items(): + action_type = str(mapping.get("type", "")) + if not action_type.startswith("UniLabJsonCommand") or action_name == "auto-close": + continue + schema = mapping.get("schema") or {} + key = f"{device_id}/{action_name}" + host_info["business_actions"][key] = { + "device_id": device_id, + "action_name": action_name, + "action_type": action_type, + "description": schema.get("description", ""), + } + # 获取设备状态 host_info["device_status"] = host_node.device_status diff --git a/unilabos/app/ws_client.py b/unilabos/app/ws_client.py index 76f6da59e..f03e4e223 100644 --- a/unilabos/app/ws_client.py +++ b/unilabos/app/ws_client.py @@ -16,7 +16,7 @@ import asyncio import traceback import websockets -import ssl as ssl_module +from websockets.exceptions import ConnectionClosed, InvalidStatus import copy from queue import Queue, Empty from dataclasses import dataclass, field @@ -33,6 +33,7 @@ from unilabos.app.communication import BaseCommunicationClient from unilabos.config.config import WSConfig, HTTPConfig, BasicConfig from unilabos.utils.log import get_comm_logger +from unilabos.app.ssl_utils import create_wss_ssl_context # 服务端通信专用 logger：独立成文件(unilabos_data/logs/ws_comm_*.log)， # 全量 TRACE 落本地、微秒级时间戳 + 线程名，便于排查通信/queue 时序问题。 @@ -456,7 +457,7 @@ async def _connection_handler(self): # 构建SSL上下文 ssl_context = None if self.websocket_url.startswith("wss://"): - ssl_context = ssl_module.create_default_context() + ssl_context = create_wss_ssl_context() ws_logger = logging.getLogger("websockets.client") ws_logger.setLevel(logging.INFO) @@ -504,13 +505,13 @@ async def _connection_handler(self): except asyncio.CancelledError: pass - except websockets.exceptions.ConnectionClosed: + except ConnectionClosed: logger.warning("[MessageProcessor] 与服务端连接中断") except TimeoutError: logger.warning( f"[MessageProcessor] 与服务端连接通信超时 (已尝试 {self.reconnect_count + 1} 次)，请检查您的网络状况" ) - except websockets.exceptions.InvalidStatus as e: + except InvalidStatus as e: logger.warning( f"[MessageProcessor] 收到服务端注册码 {e.response.status_code}, 上一进程可能还未退出" ) diff --git a/unilabos/utils/requirements.txt b/unilabos/utils/requirements.txt index 75ed4ca8a..c9944ac32 100644 --- a/unilabos/utils/requirements.txt +++ b/unilabos/utils/requirements.txt @@ -1,6 +1,7 @@ networkx typing_extensions websockets +certifi msgcenterpy>=0.1.8 orjson>=3.11 opentrons_shared_data @@ -17,4 +18,4 @@ pandas crcmod-plus pymodbus matplotlib -pylibftdi \ No newline at end of file +pylibftdi