Add custom serial class to scpi submodule

CITATION.cff

@@ -30,5 +30,5 @@ keywords:
   - single-dish
   - submillimeter
 license: MIT
-version: 0.5.0
-date-released: '2026-03-09'
+version: 0.6.0
+date-released: '2026-03-11'

docs/_static/switcher.json

@@ -1,9 +1,14 @@
 [
     {
-        "name": "0.5.0 (latest)",
-        "version": "0.5.0",
+        "name": "0.6.0 (latest)",
+        "version": "0.6.0",
         "url": "https://finerreceiver.github.io/drs4/"
     },
+    {
+        "name": "0.5.0",
+        "version": "0.5.0",
+        "url": "https://finerreceiver.github.io/drs4/0.5.0/"
+    },
     {
         "name": "0.4.0",
         "version": "0.4.0",

docs/build

@@ -5,4 +5,4 @@ export LC_ALL=C.UTF-8
 
 sphinx-apidoc -efMT -d 2 -o docs/_apidoc drs4
 sphinx-build -a -b html docs docs/_build
-sphinx-build -a -b html docs docs/_build/0.5.0
+sphinx-build -a -b html docs docs/_build/0.6.0

docs/conf.py

@@ -18,6 +18,6 @@
     ],
     "switcher": {
         "json_url": "https://finerreceiver.github.io/drs4/_static/switcher.json",
-        "version_match": "0.5.0",
+        "version_match": "0.6.0",
     },
 }

drs4/__init__.py

@@ -1,5 +1,5 @@
 __all__ = ["ctrl", "daq", "qlook", "obs", "specs", "utils"]
-__version__ = "0.5.0"
+__version__ = "0.6.0"
 
 
 # standard library

drs4/ctrl/scpi.py

@@ -3,14 +3,16 @@
 
 # standard library
 from logging import getLogger
+from serial import Serial
 from socket import socket, AF_INET, SOCK_STREAM
-from typing import IO, Sequence
+from typing import Any, IO, Sequence, overload
 
 # dependencies
 from ..utils import StrPath
 
 # constants
 DEFAULT_AUTORECV: bool = True
+DEFAULT_BAUDRATE: int = 9600
 DEFAULT_BUFSIZE: int = 4096
 DEFAULT_ENCODING: str = "ascii"
 DEFAULT_END: str = "\n"
@@ -19,6 +21,44 @@
 LOGGER = getLogger(__name__)
 
 
+class CustomSerial(Serial):
+    """Custom serial class to send/recv string with logging."""
+
+    def send(
+        self,
+        string: str,
+        flags: int = DEFAULT_FLAGS,
+        end: str = DEFAULT_END,
+        encoding: str = DEFAULT_ENCODING,
+    ) -> int:
+        """Same as Serial.write(), but accepts string, not bytes."""
+        encoded = (string + end).encode(encoding)
+
+        if (n_bytes := self.write(encoded)) is None:
+            raise ConnectionError("Unexpected disconnect.")
+
+        LOGGER.debug(f"{self.port} <- {string}")
+        return n_bytes
+
+    def recv(
+        self,
+        bufsize: int = DEFAULT_BUFSIZE,
+        flags: int = DEFAULT_FLAGS,
+        end: str = DEFAULT_END,
+        encoding: str = DEFAULT_ENCODING,
+    ) -> str:
+        """Same as Serial.read_until(), but returns string, not bytes."""
+        bend = end.encode(encoding)
+        received = self.read_until(expected=bend)
+
+        if not received or not received.endswith(bend):
+            raise TimeoutError("Unexpected disconnect.")
+
+        string = received.decode(encoding).removesuffix(end)
+        LOGGER.debug(f"{self.port} -> {string}")
+        return string
+
+
 class CustomSocket(socket):
     """Custom socket class to send/recv string with logging."""
 
@@ -63,13 +103,29 @@ def recv(
         return string
 
 
+@overload
+def connect(
+    host: None,
+    port: str,
+    /,
+    *,
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> CustomSerial: ...
+@overload
 def connect(
     host: str,
     port: int,
     /,
     *,
     timeout: float | None = DEFAULT_TIMEOUT,
-) -> CustomSocket:
+) -> CustomSocket: ...
+def connect(
+    host: Any,
+    port: Any,
+    /,
+    *,
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> Any:
     """Connect to an SCPI server and returns a custom socket object.
 
     Args:
@@ -100,12 +156,35 @@ def connect(
 
     LOGGER.debug(")")
 
-    sock = CustomSocket(AF_INET, SOCK_STREAM)
-    sock.settimeout(timeout)
-    sock.connect((host, port))
-    return sock
+    if host is None and isinstance(port, str):
+        return CustomSerial(
+            port,
+            baudrate=DEFAULT_BAUDRATE,
+            timeout=timeout,
+        )
+
+    if host is not None and isinstance(port, int):
+        conn = CustomSocket(AF_INET, SOCK_STREAM)
+        conn.settimeout(timeout)
+        conn.connect((host, port))
+        return conn
+
+    raise ValueError("Invalid host or port.")
 
 
+@overload
+def send_commands(
+    commands: IO[str] | Sequence[str] | str,
+    /,
+    *,
+    host: None,
+    port: str,
+    timeout: float | None = DEFAULT_TIMEOUT,
+    encoding: str = DEFAULT_ENCODING,
+    autorecv: bool = DEFAULT_AUTORECV,
+    bufsize: int = DEFAULT_BUFSIZE,
+) -> tuple[str, ...]: ...
+@overload
 def send_commands(
     commands: IO[str] | Sequence[str] | str,
     /,
@@ -116,6 +195,17 @@ def send_commands(
     encoding: str = DEFAULT_ENCODING,
     autorecv: bool = DEFAULT_AUTORECV,
     bufsize: int = DEFAULT_BUFSIZE,
+) -> tuple[str, ...]: ...
+def send_commands(
+    commands: IO[str] | Sequence[str] | str,
+    /,
+    *,
+    host: Any,
+    port: Any,
+    timeout: float | None = DEFAULT_TIMEOUT,
+    encoding: str = DEFAULT_ENCODING,
+    autorecv: bool = DEFAULT_AUTORECV,
+    bufsize: int = DEFAULT_BUFSIZE,
 ) -> tuple[str, ...]:
     """Send SCPI command(s) to a server.
 
@@ -152,21 +242,34 @@ def send_commands(
     if isinstance(commands, str):
         commands = (commands,)
 
-    with connect(host, port, timeout=timeout) as sock:
+    with connect(host, port, timeout=timeout) as conn:
         messages: list[str] = []
 
         for command in commands:
             if not command or command.startswith("#"):
                 continue
 
-            sock.send(command.strip(), encoding=encoding)
+            conn.send(command.strip(), encoding=encoding)
 
             if autorecv and "?" in command:
-                messages.append(sock.recv(bufsize))
+                messages.append(conn.recv(bufsize))
 
         return tuple(messages)
 
 
+@overload
+def send_commands_in(
+    path: StrPath,
+    /,
+    *,
+    host: None,
+    port: str,
+    timeout: float | None = DEFAULT_TIMEOUT,
+    encoding: str = DEFAULT_ENCODING,
+    autorecv: bool = DEFAULT_AUTORECV,
+    bufsize: int = DEFAULT_BUFSIZE,
+) -> tuple[str, ...]: ...
+@overload
 def send_commands_in(
     path: StrPath,
     /,
@@ -177,6 +280,17 @@ def send_commands_in(
     encoding: str = DEFAULT_ENCODING,
     autorecv: bool = DEFAULT_AUTORECV,
     bufsize: int = DEFAULT_BUFSIZE,
+) -> tuple[str, ...]: ...
+def send_commands_in(
+    path: StrPath,
+    /,
+    *,
+    host: Any,
+    port: Any,
+    timeout: float | None = DEFAULT_TIMEOUT,
+    encoding: str = DEFAULT_ENCODING,
+    autorecv: bool = DEFAULT_AUTORECV,
+    bufsize: int = DEFAULT_BUFSIZE,
 ) -> tuple[str, ...]:
     """Send SCPI command(s) written in a file to a server.
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "drs4"
-version = "0.5.0"
+version = "0.6.0"
 description = "Control and data acquisition software for FINER/DRS4"
 readme = "README.md"
 keywords = [
@@ -21,6 +21,7 @@ dependencies = [
     "matplotlib>=3,<4",
     "numpy>=2,<3",
     "pillow>=10,<11",
+    "pyserial>=3,<4",
     "tqdm>=4,<5",
     "typing-extensions>=4,<5",
     "xarray>=2024,<2027",