Remove the serial handler decorator (#513)

Remove the serial handler decorator

j5/backends/hardware/j5/serial.py

@@ -1,36 +1,14 @@
 """Abstract hardware backend implementation provided by j5 for serial comms."""
 from abc import abstractmethod
 from datetime import timedelta
-from functools import wraps
-from typing import Callable, Optional, Set, Type, TypeVar
+from typing import Optional, Set, Type
 
 from serial import Serial, SerialException, SerialTimeoutException
 from typing_extensions import Protocol
 
 from j5.backends import BackendMeta, CommunicationError
 from j5.boards import Board
 
-RT = TypeVar("RT")  # pragma: nocover
-
-
-def handle_serial_error(func: Callable[..., RT]) -> Callable[..., RT]:  # type: ignore
-    """
-    Wrap functions that use the serial port, and rethrow the errors.
-
-    This is a decorator that should be used to wrap any functions that call the serial
-    interface. It will catch and rethrow the errors as a CommunicationError, so that it
-    is more explicit what is going wrong.
-    """
-    @wraps(func)
-    def catch_exceptions(*args, **kwargs):  # type: ignore
-        try:
-            return func(*args, **kwargs)
-        except SerialTimeoutException as e:
-            raise CommunicationError(f"Serial Timeout Error: {e}") from e
-        except SerialException as e:
-            raise CommunicationError(f"Serial Error: {e}") from e
-    return catch_exceptions
-
 
 class Seriallike(Protocol):
     """
@@ -68,7 +46,6 @@ def write(self, data: bytes) -> int:
 class SerialHardwareBackend(metaclass=BackendMeta):
     """An abstract class for creating backends that use USB serial communication."""
 
-    @handle_serial_error
     def __init__(
             self,
             serial_port: str,
@@ -77,11 +54,16 @@ def __init__(
             timeout: timedelta = timedelta(milliseconds=250),
     ) -> None:
         timeout_secs = timeout / timedelta(seconds=1)
-        self._serial = serial_class(
-            port=serial_port,
-            baudrate=baud,
-            timeout=timeout_secs,
-        )
+        try:
+            self._serial = serial_class(
+                port=serial_port,
+                baudrate=baud,
+                timeout=timeout_secs,
+            )
+        except SerialTimeoutException as e:
+            raise CommunicationError(f"Serial Timeout Error: {e}") from e
+        except SerialException as e:
+            raise CommunicationError(f"Serial Error: {e}") from e
 
     @classmethod
     @abstractmethod
@@ -95,10 +77,14 @@ def firmware_version(self) -> Optional[str]:
         """The firmware version of the board."""
         raise NotImplementedError  # pragma: no cover
 
-    @handle_serial_error
     def read_serial_line(self, empty: bool = False) -> str:
         """Read a line from the serial interface."""
-        bdata = self._serial.readline()
+        try:
+            bdata = self._serial.readline()
+        except SerialTimeoutException as e:
+            raise CommunicationError(f"Serial Timeout Error: {e}") from e
+        except SerialException as e:
+            raise CommunicationError(f"Serial Error: {e}") from e
 
         if len(bdata) == 0:
             if empty:

j5/backends/hardware/sb/arduino.py

@@ -4,16 +4,13 @@
 from threading import Lock
 from typing import Callable, List, Mapping, Optional, Set, Tuple, Type
 
-from serial import Serial
+from serial import Serial, SerialException, SerialTimeoutException
 from serial.tools.list_ports import comports
 from serial.tools.list_ports_common import ListPortInfo
 
 from j5.backends import CommunicationError
 from j5.backends.hardware.env import NotSupportedByHardwareError
-from j5.backends.hardware.j5.serial import (
-    SerialHardwareBackend,
-    handle_serial_error,
-)
+from j5.backends.hardware.j5.serial import SerialHardwareBackend
 from j5.boards import Board
 from j5.boards.sb.arduino import SBArduinoBoard
 from j5.components import GPIOPinInterface, GPIOPinMode, LEDInterface
@@ -74,14 +71,17 @@ def discover(
             boards.add(
                 SBArduinoBoard(
                     port.serial_number,
-                    cls(port.device, serial_class),
+                    cls(port.device, serial_class),  # type: ignore
                 ),
             )
 
         return boards
 
-    @handle_serial_error
-    def __init__(self, serial_port: str, serial_class: Type[Serial] = Serial) -> None:
+    def __init__(
+            self,
+            serial_port: str,
+            serial_class: Type[Serial] = Serial,
+    ) -> None:
         super(SBArduinoHardwareBackend, self).__init__(
             serial_port=serial_port,
             serial_class=serial_class,
@@ -131,29 +131,33 @@ def firmware_version(self) -> Optional[str]:
         """The firmware version of the board."""
         return self._version_line.split("v")[1]
 
-    @handle_serial_error
     def _command(self, command: str, *params: str) -> List[str]:
         """Send a command to the board."""
-        with self._lock:
-            message = " ".join([command] + list(params)) + "\n"
-            self._serial.write(message.encode("utf-8"))
-
-            results: List[str] = []
-            while True:
-                line = self.read_serial_line(empty=False)
-                code, param = line.split(None, 1)
-                if code == "+":
-                    return results
-                elif code == "-":
-                    raise CommunicationError(f"Arduino error: {param}")
-                elif code == ">":
-                    results.append(param)
-                elif code == "#":
-                    pass  # Ignore comment lines
-                else:
-                    raise CommunicationError(
-                        f"Arduino returned unrecognised response line: {line}",
-                    )
+        try:
+            with self._lock:
+                message = " ".join([command] + list(params)) + "\n"
+                self._serial.write(message.encode("utf-8"))
+
+                results: List[str] = []
+                while True:
+                    line = self.read_serial_line(empty=False)
+                    code, param = line.split(None, 1)
+                    if code == "+":
+                        return results
+                    elif code == "-":
+                        raise CommunicationError(f"Arduino error: {param}")
+                    elif code == ">":
+                        results.append(param)
+                    elif code == "#":
+                        pass  # Ignore comment lines
+                    else:
+                        raise CommunicationError(
+                            f"Arduino returned unrecognised response line: {line}",
+                        )
+        except SerialTimeoutException as e:
+            raise CommunicationError(f"Serial Timeout Error: {e}") from e
+        except SerialException as e:
+            raise CommunicationError(f"Serial Error: {e}") from e
 
     def _update_digital_pin(self, identifier: int) -> None:
         if identifier >= FIRST_ANALOGUE_PIN:

j5/backends/hardware/sr/v4/motor_board.py

@@ -2,16 +2,12 @@
 from threading import Lock
 from typing import Callable, List, Optional, Set, Type, cast
 
-from serial import Serial
+from serial import Serial, SerialException, SerialTimeoutException
 from serial.tools.list_ports import comports
 from serial.tools.list_ports_common import ListPortInfo
 
-from j5.backends import CommunicationError
-from j5.backends.hardware.j5.serial import (
-    SerialHardwareBackend,
-    Seriallike,
-    handle_serial_error,
-)
+from j5.backends import Backend, CommunicationError
+from j5.backends.hardware.j5.serial import SerialHardwareBackend, Seriallike
 from j5.boards import Board
 from j5.boards.sr.v4.motor_board import MotorBoard
 from j5.components.motor import MotorInterface, MotorSpecialState, MotorState
@@ -55,13 +51,15 @@ def discover(
             boards.add(
                 MotorBoard(
                     port.serial_number,
-                    SRV4MotorBoardHardwareBackend(port.device, serial_class),
+                    cast(
+                        Backend,
+                        SRV4MotorBoardHardwareBackend(port.device, serial_class),
+                    ),
                 ),
             )
 
         return boards
 
-    @handle_serial_error
     def __init__(self, serial_port: str, serial_class: Type[Seriallike] = Serial) -> None:
         super(SRV4MotorBoardHardwareBackend, self).__init__(
             serial_port=serial_port,
@@ -98,25 +96,34 @@ def __del__(self) -> None:
                     MotorSpecialState.BRAKE,
                     acquire_lock=False,
                 )
-        self._serial.flush()
-        self._serial.close()
+        try:
+            self._serial.flush()
+            self._serial.close()
+        except SerialTimeoutException as e:
+            raise CommunicationError(f"Serial Timeout Error: {e}") from e
+        except SerialException as e:
+            raise CommunicationError(f"Serial Error: {e}") from e
 
-    @handle_serial_error
     def send_command(self, command: int, data: Optional[int] = None) -> None:
         """Send a serial command to the board."""
         with self._lock:
             return self._send_command_no_lock(command, data)
 
     def _send_command_no_lock(self, command: int, data: Optional[int] = None) -> None:
         """Send a serial command to the board without acquiring the lock."""
-        message: List[int] = [command]
-        if data is not None:
-            message += [data]
-        bytes_written = self._serial.write(bytes(message))
-        if len(message) != bytes_written:
-            raise CommunicationError(
-                "Mismatch in command bytes written to serial interface.",
-            )
+        try:
+            message: List[int] = [command]
+            if data is not None:
+                message += [data]
+            bytes_written = self._serial.write(bytes(message))
+            if len(message) != bytes_written:
+                raise CommunicationError(
+                    "Mismatch in command bytes written to serial interface.",
+                )
+        except SerialTimeoutException as e:
+            raise CommunicationError(f"Serial Timeout Error: {e}") from e
+        except SerialException as e:
+            raise CommunicationError(f"Serial Error: {e}") from e
 
     @property
     def firmware_version(self) -> Optional[str]:

j5/boards/sb/arduino.py

@@ -37,7 +37,11 @@ class SBArduinoBoard(Board):
     _analogue_pins: Mapping[AnaloguePin, GPIOPin]
     name: str = "Arduino Uno"
 
-    def __init__(self, serial: str, backend: Backend):
+    def __init__(
+            self,
+            serial: str,
+            backend: Backend,
+    ):
         self._serial = serial
         self._backend = backend