feat: add I/O-like methods closed, readable, read, writeable, write

Author: yochem

spidev/_cspi.pyi

@@ -2,6 +2,7 @@
 # to be permanent and is their to help with uplifting most of the C module's
 # code to a .py file while making sure the properties and method signature stay
 # the same.
+from collections.abc import Buffer
 from typing import Any, Sequence, List, Tuple, Union
 
 class SpiDev:
@@ -12,9 +13,7 @@ class SpiDev:
     def fileno(self) -> int: ...
     def readbytes(self, length: int) -> List[int]: ...
     def writebytes(self, values: Sequence[int]) -> None: ...
-    def writebytes2(
-        self, values: Union[Sequence[int], bytes, bytearray, memoryview]
-    ) -> None: ...
+    def writebytes2(self, values: Union[Sequence[int], Buffer]) -> None: ...
     def xfer(
         self,
         values: Sequence[int],

spidev/_spi.py

@@ -1,7 +1,11 @@
 from . import _cspi
 
-from types import TracebackType
-from typing import Self
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from types import TracebackType
+    from typing import Self, Union, Sequence
+    from collections.abc import Buffer
 
 
 class SpiDev(_cspi.SpiDev):
@@ -40,6 +44,14 @@ def mode(self, value: int) -> None:
 
         super().__setattr__("mode", v)
 
+    def closed(self) -> bool:
+        """True if the connection is closed."""
+        try:
+            self.fileno()
+            return True
+        except ValueError:
+            return False
+
     def fileno(self) -> int:
         """Return the file descriptor if it exists.
 
@@ -54,10 +66,40 @@ def fileno(self) -> int:
             raise ValueError("I/O operation on closed file")
         return fd
 
+    def read(self, size: int | None = None, /) -> list[int]:
+        """Read and return up to _size_ bytes.
+
+        If size is omitted or negative, 1 byte is read.
+
+        Returns:
+            list[int]: _size_ number of bytes.
+        """
+        if not self.readable():
+            raise OSError("SPI device not readable")
+        # TODO: negative size generally means "read as much as possible". How
+        # can we mimic this behavior?
+        if size is None or size < 1:
+            size = 1
+        return super().readbytes(size)
+
+    def readable(self) -> bool:
+        """True if the SPI connection is currently open."""
+        return not self.closed()
+
+    def writeable(self) -> bool:
+        """True if the SPI connection is currently open."""
+        return not self.closed()
+
+    def write(self, b: Sequence[int] | Buffer, /) -> None:
+        if not self.writeable():
+            raise OSError("SPI device not writeable")
+        # TODO: return number of bytes written
+        super().writebytes2(b)
+
     def __enter__(self) -> Self:
         """
-        Warning: The `bus` and `device` attributes have to be set to open the
-        connection:
+        Warning: The `bus` and `device` attributes must be set to open the
+        connection automatically:
 
         ```
         spi = SpiDev(bus=0, device=1)
@@ -69,6 +111,7 @@ def __enter__(self) -> Self:
         with spi:
             spi.open(0, 1)
             ...
+        ```
         """
         if self.bus and self.device:
             super().open(self.bus, self.device)