Format code using black and isort

Author: basilfx

tests/test_tinylink.py

@@ -1,4 +1,5 @@
 import unittest
+
 import tinylink
 
 
@@ -17,7 +18,7 @@ def __init__(self) -> None:
         self.length = 0
 
     def read(self, size: int) -> bytes:
-        data = self.stream[self.index:min(self.length, self.index + size)]
+        data = self.stream[self.index : min(self.length, self.index + size)]
         self.index += len(data)
 
         # Return data.
@@ -48,8 +49,12 @@ def test_basic(self):
         size = link.write(message)
 
         self.assertEqual(
-            size, tinylink.LEN_PREAMBLE + tinylink.LEN_HEADER +
-            tinylink.LEN_BODY + len(message))
+            size,
+            tinylink.LEN_PREAMBLE
+            + tinylink.LEN_HEADER
+            + tinylink.LEN_BODY
+            + len(message),
+        )
 
         # Read `size` bytes to receive the full frame, test it partially.
         link.read(1)
@@ -155,7 +160,7 @@ def test_damaged_a(self):
         message = b"Hello, this is a test"
 
         size = link.write(message)
-        handle.stream[-tinylink.LEN_CRC:] = [0x00] * tinylink.LEN_CRC
+        handle.stream[-tinylink.LEN_CRC :] = [0x00] * tinylink.LEN_CRC
         frames = link.read(size)
 
         self.assertEqual(len(frames), 1)
@@ -173,7 +178,7 @@ def test_damaged_b(self):
         message = b"Hello, this is a test"
 
         size = link.write(message)
-        handle.stream[tinylink.LEN_PREAMBLE+tinylink.LEN_HEADER-1] = 0x00
+        handle.stream[tinylink.LEN_PREAMBLE + tinylink.LEN_HEADER - 1] = 0x00
         frames = link.read(size)
 
         self.assertEqual(len(frames), 0)

tinylink/__init__.py

@@ -1,4 +1,4 @@
 from .consts import *  # noqa
 from .link import Frame, TinyLink  # noqa
 
-__version__ = "2.0.0"
+__version__ = "2.0.0"

tinylink/cli.py

@@ -1,13 +1,13 @@
+import argparse
 import csv
-import sys
-import time
 import select
 import struct
+import sys
+import time
+from io import StringIO
 from typing import Optional
-import tinylink
-import argparse
 
-from io import StringIO
+import tinylink
 
 try:
     import serial
@@ -24,13 +24,17 @@ def parse_arguments(argv: list[str]) -> argparse.Namespace:
 
     # Add options.
     parser.add_argument("port", type=str, help="serial port")
+    parser.add_argument("baudrate", type=int, default=9600, help="serial baudrate")
     parser.add_argument(
-        "baudrate", type=int, default=9600, help="serial baudrate")
+        "--length", type=int, default=2**16, help="maximum length of frame"
+    )
     parser.add_argument(
-        "--length", type=int, default=2**16, help="maximum length of frame")
-    parser.add_argument(
-        "--endianness", type=str, default="little", choices=["big", "little"],
-        help="maximum length of frame")
+        "--endianness",
+        type=str,
+        default="little",
+        choices=["big", "little"],
+        help="maximum length of frame",
+    )
 
     # Parse command line.
     return parser.parse_args(argv[1:])
@@ -52,7 +56,7 @@ def dump(prefix: str, data: bytes) -> str:
             if i + j < length:
                 b = data[i + j]
                 hexstr += "%02x " % b
-                bytestr += bytes((b, )) if 0x20 <= b < 0x7F else b"."
+                bytestr += bytes((b,)) if 0x20 <= b < 0x7F else b"."
             else:
                 hexstr += "   "
 
@@ -126,13 +130,13 @@ def process_stdin(link: tinylink.TinyLink) -> Optional[bool]:
                 except:
                     try:
                         # Assume it is an int.
-                        value = struct.pack(
-                            link.endianness + pack, int(item, 0))
+                        value = struct.pack(link.endianness + pack, int(item, 0))
                     except ValueError:
                         # Assume it is a byte string.
                         item_bytes = item.encode("ascii")
                         value = struct.pack(
-                            link.endianness + str(len(item_bytes)) + "s", item_bytes)
+                            link.endianness + str(len(item_bytes)) + "s", item_bytes
+                        )
 
                 # Concat to frame.
                 frame.data = (frame.data or bytes()) + value
@@ -171,7 +175,8 @@ def main(argv: list[str]) -> int:
     if serial is None:
         sys.stdout.write(
             "TinyLink CLI uses PySerial, but it is not installed. Please "
-            "install this first.\n")
+            "install this first.\n"
+        )
         return 1
 
     # Parse arguments.
@@ -184,8 +189,7 @@ def main(argv: list[str]) -> int:
 
     # Open serial port and create link.
     handle = serial.Serial(arguments.port, baudrate=arguments.baudrate)
-    link = tinylink.TinyLink(
-        handle, max_length=arguments.length, endianness=endianness)
+    link = tinylink.TinyLink(handle, max_length=arguments.length, endianness=endianness)
 
     # Loop until finished.
     try:
@@ -214,6 +218,7 @@ def main(argv: list[str]) -> int:
     # Done.
     return 0
 
+
 # E.g. `python cli.py /dev/tty.usbmodem1337 --baudrate 9600`.
 if __name__ == "__main__":
     run()

tinylink/link.py

@@ -1,9 +1,7 @@
+import struct
 from typing import Protocol
 
-from . import consts
-from . import utils
-
-import struct
+from . import consts, utils
 
 
 class Handle(Protocol):
@@ -12,7 +10,7 @@ class Handle(Protocol):
     """
 
     def read(self, size: int) -> bytes:
-        """"
+        """ "
         Read up to  `size` bytes.
         """
 
@@ -31,8 +29,9 @@ class Frame:
     flags: int
     damaged: int
 
-    def __init__(self, data: bytes = None, flags: int = consts.FLAG_NONE,
-                 damaged: bool = False) -> None:
+    def __init__(
+        self, data: bytes = None, flags: int = consts.FLAG_NONE, damaged: bool = False
+    ) -> None:
         if data is not None:
             if type(data) is not bytes:
                 raise ValueError("Provided data must be encoded as bytes.")
@@ -45,7 +44,11 @@ def __init__(self, data: bytes = None, flags: int = consts.FLAG_NONE,
 
     def __repr__(self) -> str:
         return "%s(%s, flags=%d, damaged=%s)" % (
-            self.__class__.__name__, repr(self.data), self.flags, self.damaged)
+            self.__class__.__name__,
+            repr(self.data),
+            self.flags,
+            self.damaged,
+        )
 
 
 class TinyLink:
@@ -67,9 +70,13 @@ class TinyLink:
     max_length: int
     ignore_damaged: bool
 
-    def __init__(self, handle: Handle, endianness: str = consts.LITTLE_ENDIAN,
-                 max_length: int = 2**(consts.LEN_LENGTH * 8),
-                 ignore_damaged: bool = False) -> None:
+    def __init__(
+        self,
+        handle: Handle,
+        endianness: str = consts.LITTLE_ENDIAN,
+        max_length: int = 2 ** (consts.LEN_LENGTH * 8),
+        ignore_damaged: bool = False,
+    ) -> None:
         """
         Construct a new TinyLink state machine. A state machine takes a handle,
         which provides a `read` and `write` method.
@@ -115,21 +122,25 @@ def write_frame(self, frame: Frame) -> int:
         # Check length of message.
         if length > self.max_length:
             raise ValueError(
-                "Message length %d exceeds max length %d" % (
-                    length, self.max_length))
+                "Message length %d exceeds max length %d" % (length, self.max_length)
+            )
 
         # Pack header.
         checksum_header = utils.checksum_header(frame.flags, length)
         result += struct.pack(
-            self.endianness + "IHHB", consts.PREAMBLE, frame.flags, length,
-            checksum_header)
+            self.endianness + "IHHB",
+            consts.PREAMBLE,
+            frame.flags,
+            length,
+            checksum_header,
+        )
 
         # Pack data.
         if frame.data is not None:
             checksum_frame = utils.checksum_frame(frame.data, checksum_header)
             result += struct.pack(
-                self.endianness + str(length) + "sI", frame.data,
-                checksum_frame)
+                self.endianness + str(length) + "sI", frame.data, checksum_frame
+            )
 
         # Write to file.
         return self.handle.write(result)
@@ -164,14 +175,18 @@ def read(self, limit: int = 1) -> list[Frame]:
             # Decide what to do.
             if self.state == consts.WAITING_FOR_PREAMBLE:
                 if self.index >= consts.LEN_PREAMBLE:
-                    start, = struct.unpack_from(
-                        self.endianness + "I", self.buffer, self.index - 4)
+                    (start,) = struct.unpack_from(
+                        self.endianness + "I", self.buffer, self.index - 4
+                    )
 
                     if start == consts.PREAMBLE:
                         # Advance to next state.
                         self.index = 0
                         self.state = consts.WAITING_FOR_HEADER
-                    elif self.index == self.max_length + consts.LEN_HEADER + consts.LEN_BODY:
+                    elif (
+                        self.index
+                        == self.max_length + consts.LEN_HEADER + consts.LEN_BODY
+                    ):
                         # Preamble not found and stream is full. Copy last four
                         # bytes, because the next byte may form the preamble
                         # together with the last three bytes.
@@ -181,11 +196,14 @@ def read(self, limit: int = 1) -> list[Frame]:
             elif self.state == consts.WAITING_FOR_HEADER:
                 if self.index == consts.LEN_HEADER:
                     flags, length, checksum = struct.unpack_from(
-                        self.endianness + "HHB", self.buffer)
+                        self.endianness + "HHB", self.buffer
+                    )
 
                     # Verify checksum.
-                    if checksum == utils.checksum_header(flags, length) and \
-                            length <= self.max_length:
+                    if (
+                        checksum == utils.checksum_header(flags, length)
+                        and length <= self.max_length
+                    ):
 
                         if length > 0:
                             self.state = consts.WAITING_FOR_BODY
@@ -203,13 +221,16 @@ def read(self, limit: int = 1) -> list[Frame]:
             elif self.state == consts.WAITING_FOR_BODY:
                 # Unpack header.
                 flags, length, checksum_a = struct.unpack_from(
-                    self.endianness + "HHB", self.buffer)
+                    self.endianness + "HHB", self.buffer
+                )
 
                 if self.index == consts.LEN_HEADER + length + consts.LEN_CRC:
                     # Unpack body.
                     result, checksum_b = struct.unpack_from(
                         self.endianness + str(length) + "sI",
-                        self.buffer, consts.LEN_HEADER)
+                        self.buffer,
+                        consts.LEN_HEADER,
+                    )
 
                     # Verify checksum.
                     if checksum_b == utils.checksum_frame(result, checksum_a):

tinylink/utils.py

@@ -11,7 +11,7 @@ def crc32(buf):
 
     def crc32_value(c):
         ulTemp1 = (result >> 8) & 0x00FFFFFF
-        ulCRC = (result ^ c) & 0xff
+        ulCRC = (result ^ c) & 0xFF
 
         for i in range(8):
             if ulCRC & 0x01:
@@ -46,5 +46,4 @@ def checksum_frame(data, checksum_header):
     Calculate checksum of both the checksum header and the data.
     """
 
-    return crc32(
-        memoryview(data).tobytes() + bytearray([checksum_header])) & 0xFFFFFFFF
+    return crc32(memoryview(data).tobytes() + bytearray([checksum_header])) & 0xFFFFFFFF