284 refactor pulse generator severpy to inherit from abstractinstrumentserver

docs/how-to/4c_pulse_genertor_shanghai_tech.md

@@ -0,0 +1,70 @@
+# Pulse Generator Server (ShanghaiTech)
+
+A TCP/IP instrument server designed to control Pulse Generators via USB Serial (RS232).
+
+## 1. Connection Specifications
+
+| Parameter | Default Value |
+| :--- | :--- |
+| **Host** | `localhost` (127.0.0.1) |
+| **TCP Port** | `8888` |
+| **Interface** | USB Serial (COM / /dev/tty) |
+| **Baud Rate** | `9600` |
+| **Data Protocol** | Tab-Separated, Newline-Terminated (`\t`, `\n`) |
+
+---
+
+## 2. Communication Protocol
+
+
+
+The server follows a **Request-Response** model. Every command sent by a client will receive a response starting with a status bit.
+
+### Request Format
+`COMMAND` + `\t` (Tab) + `ARGUMENT` (Optional) + `\n` (Newline)
+
+### Response Format
+* **Success:** `1` + `\t` + `Data/Message` + `\n`
+* **Error:** `0` + `\t` + `Error Description` + `\n`
+
+---
+
+## 3. Command Registry
+
+| Command | Argument | Description | Example |
+| :--- | :--- | :--- | :--- |
+| `ping` | None | Heartbeat check to verify server status. | `ping\n` |
+| `connect_hardware` | None | Initializes/Re-opens the Serial port. | `connect_hardware\n` |
+| `set_delay` | `0-1023` | Sets the pulse delay on the hardware. | `set_delay\t512\n` |
+| `get_delay` | None | Queries the current delay from hardware. | `get_delay\n` |
+| `reset_serial_buffer`| None | Clears the hardware's internal I/O buffers. | `reset_serial_buffer\n` |
+| `pass_command` | `string` | Sends a raw AT command to the device. | `pass_command\tAT+VER\n` |
+| `shutdown` | None | Safely stops the server and releases hardware. | `shutdown\n` |
+
+---
+
+
+## 4. Quick Start: Python Client
+
+You can interact with the server using any language that supports sockets. Here is a minimal Python example:
+
+```python
+import socket
+
+def send_pulse_command(ip, port, cmd, arg=None):
+    try:
+        with socket.create_connection((ip, port), timeout=2.0) as s:
+            message = f"{cmd}\t{arg}\n" if arg else f"{cmd}\n"
+            s.sendall(message.encode())
+            response = s.recv(1024).decode().strip()
+
+            status, data = response.split('\t', 1)
+            if status == '1':
+                print(f"SUCCESS: {data}")
+            else:
+                print(f"ERROR: {data}")
+    except Exception as e:
+        print(f"Connection Failed: {e}")
+
+# Example Usage
+send_pulse_command("127.0.0.1", 8888, "set_delay", "250")

pyproject.toml

@@ -21,14 +21,19 @@ dependencies = [
     "dls-dodal>=2.2.0",
     "ophyd-async[sim]",
     "scanspec",
+
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"
 requires-python = ">=3.11"
 
+
+[project.optional-dependencies]
+server = ["pyserial"]
 [dependency-groups]
 dev = [
+    "sm_bluesky[server]",
     "copier",
     "myst-parser",
     "pre-commit",

src/sm_bluesky/common/server/__init__.py

@@ -1,3 +1,4 @@
 from .abstract_instrument_server import AbstractInstrumentServer
+from .pulse_generator_shanghai_tech import GeneratorServerShanghaiTech
 
-__all__ = ["AbstractInstrumentServer"]
+__all__ = ["AbstractInstrumentServer", "GeneratorServerShanghaiTech"]

src/sm_bluesky/common/server/pulse_generator_shanghai_tech.py

@@ -0,0 +1,102 @@
+import logging
+
+from serial import Serial
+
+from sm_bluesky.common.server import AbstractInstrumentServer
+from sm_bluesky.log import LOGGER
+
+
+class GeneratorServerShanghaiTech(AbstractInstrumentServer):
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        ipv6: bool = False,
+        usb_port: str = "COM4",
+        baud_rate: int = 9600,
+        timeout: float = 1.0,
+        max_pulse_delay=1024,
+    ):
+        super().__init__(host, port, ipv6)
+        self.usb_port: str = usb_port
+        self.baud_rate: int = baud_rate
+        self.timeout: float = timeout
+        self.max_pulse_delay: float = max_pulse_delay
+        self.device: Serial | None = None
+
+        # Expand the registry with Pulse Generator specific commands
+        self._command_registry.update(
+            {
+                b"set_delay": self._set_delay,
+                b"get_delay": self._get_delay,
+                b"reset_serial_buffer": self._reset_serial_buffer,
+                b"pass_command": self._passthrough,
+            }
+        )
+
+    def connect_hardware(self) -> bool:
+        """Initialize the USB connection protocol."""
+        try:
+            self.device = Serial(
+                port=self.usb_port, baudrate=self.baud_rate, timeout=self.timeout
+            )
+            self._send_response(b"Hardware connected successfully")
+            return True
+        except Exception as e:
+            self._error_helper(message="Failed to connect to hardware", error=e)
+            return False
+
+    def disconnect_hardware(self):
+        """Safely release the USB resource."""
+        if self.device and self.device.is_open:
+            try:
+                self.device.close()
+            except Exception as e:
+                self._error_helper(
+                    message="Error occurred while closing hardware connection", error=e
+                )
+            self._hardware_connected = False
+            self.device = None
+            LOGGER.info("Hardware disconnected successfully")
+            self._send_response(b"Hardware disconnected")
+        else:
+            self._error_helper(
+                message="Attempted to disconnect hardware that was not connected",
+                level=logging.WARNING,
+            )
+
+    def _set_delay(self, value: bytes) -> None:
+        delay = int(value.decode("utf-8"))
+        if self.max_pulse_delay > delay >= 0:
+            self._send_hardware_command(b"AT+DLSET=" + value)
+            LOGGER.info(f"Setting delay to {value}")
+        else:
+            raise ValueError(
+                f"Delay {delay} is out of bounds (0-{self.max_pulse_delay - 1})"
+            )
+
+    def _get_delay(self):
+        self._send_hardware_command(b"AT+DLSET=?")
+        LOGGER.info("Reading delay")
+
+    def _reset_serial_buffer(self):
+        if not self.device:
+            raise ConnectionError(
+                "Hardware not connected. Call connect_hardware first."
+            )
+        self.device.reset_input_buffer()
+        self.device.reset_output_buffer()
+        LOGGER.info("Reseting buffers")
+
+    def _passthrough(self, value: bytes):
+        self._send_hardware_command(value)
+
+    def _send_hardware_command(self, cmd: bytes) -> None:
+        if not self.device:
+            raise ConnectionError(
+                "Hardware not connected. Call connect_hardware first."
+            )
+        self.device.write(cmd + b"\r\n")
+        self.device.flush()
+        device_respond = self.device.readline().strip()
+        self._send_response(device_respond)