lint

miniscope_io/devices/__init__.py

@@ -1,3 +1,3 @@
 """
 Control interfaces for external hardware devices
-"""
+"""

miniscope_io/devices/opalkelly.py

@@ -2,9 +2,13 @@
 Interfaces for OpalKelly (model number?) FPGAs
 """
 
-from miniscope_io.vendor import opalkelly as ok
-from miniscope_io.exceptions import StreamReadError, DeviceConfigurationError, DeviceOpenError
+from miniscope_io.exceptions import (
+    DeviceConfigurationError,
+    DeviceOpenError,
+    StreamReadError,
+)
 from miniscope_io.logging import init_logger
+from miniscope_io.vendor import opalkelly as ok
 
 
 class okDev(ok.okCFrontPanel):
@@ -20,7 +24,7 @@ class okDev(ok.okCFrontPanel):
 
     def __init__(self, serial_id: str = ""):
         super().__init__()
-        self.logger = init_logger('okDev')
+        self.logger = init_logger("okDev")
         ret = self.OpenBySerial("")
         if ret != self.NoError:
             raise DeviceOpenError(f"Cannot open device: {serial_id}")
@@ -41,7 +45,9 @@ def uploadBit(self, bit_file: str) -> None:
         if ret == self.NoError:
             self.logger.debug(f"Succesfully uploaded {bit_file}")
         else:
-            raise DeviceConfigurationError(f"Configuration of {self.info.productName} failed")
+            raise DeviceConfigurationError(
+                f"Configuration of {self.info.productName} failed"
+            )
         self.logger.debug(
             "FrontPanel {} supported".format(
                 "is" if self.IsFrontPanelEnabled() else "not"

miniscope_io/exceptions.py

@@ -2,6 +2,7 @@
 Custom exceptions!
 """
 
+
 class InvalidSDException(Exception):
     """
     Raised when :class:`.io.SDCard` is used with a drive that doesn't have the
@@ -14,26 +15,31 @@ class EndOfRecordingException(StopIteration):
     Raised when :class:`.io.SDCard` is at the end of the available recording!
     """
 
+
 class StreamError(RuntimeError):
     """
     Base class for errors while streaming data
     """
 
+
 class StreamReadError(StreamError):
     """
     Error while reading streaming data from a device
     """
 
+
 class DeviceError(RuntimeError):
     """
     Base class for errors when communicating with or configuring devices
     """
 
+
 class DeviceOpenError(DeviceError):
     """
     Error opening a connection to a device
     """
 
+
 class DeviceConfigurationError(DeviceError):
     """
     Error while configuring a device

miniscope_io/io.py

@@ -171,7 +171,7 @@ def frame_count(self) -> int:
             warnings.warn(
                 "Got more frames than indicated in card header, expected "
                 f"{self._frame_count} but got {max_pos}",
-                stacklevel=1
+                stacklevel=1,
             )
             self._frame_count = int(max_pos)
 
@@ -194,7 +194,7 @@ def __enter__(self) -> "SDCard":
         self._last_buffer_n = 0
         self._frame = 0
 
-        self._f = open(self.drive, "rb") # noqa: SIM115 - this is a context handler
+        self._f = open(self.drive, "rb")  # noqa: SIM115 - this is a context handler
         # seek to the start of the data
         self._f.seek(self.layout.sectors.data_pos, 0)
         # store the 0th frame position
@@ -288,7 +288,7 @@ def _trim(self, data: np.ndarray, expected_size: int) -> np.ndarray:
             warnings.warn(
                 f"Expected buffer data length: {expected_size}, got data with shape "
                 f"{data.shape}. Padding to expected length",
-                stacklevel=1
+                stacklevel=1,
             )
 
             # trim if too long
@@ -340,14 +340,18 @@ def read(self, return_header: bool = False) -> Union[np.ndarray, Frame]:
                     # blank,  and thus have a value of 0 for the header size, and we
                     # can't read 0 from the card.
                     self._f.seek(last_position, 0)
-                    raise EndOfRecordingException("Reached the end of the video!") from None
+                    raise EndOfRecordingException(
+                        "Reached the end of the video!"
+                    ) from None
                 else:
                     raise e
             except IndexError as e:
                 if "index 0 is out of bounds for axis 0 with size 0" in str(e):
                     # end of file if we are reading from a disk image without any
                     # additional space on disk
-                    raise EndOfRecordingException("Reached the end of the video!") from None
+                    raise EndOfRecordingException(
+                        "Reached the end of the video!"
+                    ) from None
                 else:
                     raise e
 
@@ -378,8 +382,8 @@ def read(self, return_header: bool = False) -> Union[np.ndarray, Frame]:
 
     def to_video(
         self,
-        path: Union[Path,str],
-        fourcc: Literal['GREY', 'mp4v', 'XVID'] = 'GREY',
+        path: Union[Path, str],
+        fourcc: Literal["GREY", "mp4v", "XVID"] = "GREY",
         isColor: bool = False,
         force: bool = False,
         progress: bool = True,

miniscope_io/models/config.py

@@ -85,7 +85,7 @@ def folder_exists(cls, v: Path) -> Path:
         v = Path(v)
         v.mkdir(exist_ok=True, parents=True)
 
-        assert v.exists(), f'{v} does not exist!'
+        assert v.exists(), f"{v} does not exist!"
         return v
 
     @model_validator(mode="after")

miniscope_io/models/stream.py

@@ -3,7 +3,7 @@
 """
 
 from pathlib import Path
-from typing import Optional, Union, Literal
+from typing import Literal, Optional, Union
 
 from pydantic import field_validator
 
@@ -80,7 +80,7 @@ class StreamDaqConfig(MiniscopeConfig, YAMLMixin):
         Takuya - double-check the definitions around blocks and buffers in the firmware and add description.
     """
 
-    mode: Literal["DAQ", "RAW_RECORD", "RAW_REPLAY"] = 'DAQ'
+    mode: Literal["DAQ", "RAW_RECORD", "RAW_REPLAY"] = "DAQ"
     device: str
     bitstream: Optional[Path]
     port: Optional[str]

miniscope_io/stream_daq.py

@@ -1,37 +1,38 @@
 import argparse
 import multiprocessing
-import sys
 import os
+import sys
 import time
-from typing import Literal, Optional, Tuple, List
 from pathlib import Path
+from typing import List, Literal, Optional, Tuple
 
 import cv2
 import numpy as np
 import serial
 from bitstring import Array, BitArray, Bits
 
 from miniscope_io import init_logger
-
+from miniscope_io.exceptions import EndOfRecordingException, StreamReadError
 from miniscope_io.formats.stream import StreamBufferHeader
 from miniscope_io.models.buffer import BufferHeader
 from miniscope_io.models.stream import StreamBufferHeaderFormat, StreamDaqConfig
-from miniscope_io.exceptions import EndOfRecordingException, StreamReadError
 from tests.mock.opalkelly import okDevMock
 
 HAVE_OK = False
 ok_error = None
 try:
     from miniscope_io.devices.opalkelly import okDev
+
     HAVE_OK = True
 except (ImportError, ModuleNotFoundError):
     module_logger = init_logger("streamDaq")
     module_logger.warning(
-        "Could not import OpalKelly driver, you can't read from FPGA!\nCheck out Opal Kelly's website for install info\nhttps://docs.opalkelly.com/fpsdk/getting-started/")
+        "Could not import OpalKelly driver, you can't read from FPGA!\nCheck out Opal Kelly's website for install info\nhttps://docs.opalkelly.com/fpsdk/getting-started/"
+    )
 
 # Parsers for daq inputs
 daqParser = argparse.ArgumentParser("streamDaq")
-daqParser.add_argument("-c", "--config", help='YAML config file path: string')
+daqParser.add_argument("-c", "--config", help="YAML config file path: string")
 
 
 class StreamDaq:
@@ -78,7 +79,7 @@ def __init__(
         self.preamble = self.config.preamble
         self._buffer_npix: Optional[List[int]] = None
         self._nbuffer_per_fm: Optional[int] = None
-        self.terminate = multiprocessing.Value('b', False)
+        self.terminate = multiprocessing.Value("b", False)
 
     @property
     def buffer_npix(self) -> List[int]:
@@ -256,9 +257,9 @@ def _fpga_recv(
                 self.terminate.value = True
                 break
 
-            if self.config.mode == 'RAW_RECORD':
-                fpga_raw_path = 'data/fpga_raw.bin' # Not sure where to define this because we don't want to overwrite.
-                with open(fpga_raw_path, 'wb') as file:
+            if self.config.mode == "RAW_RECORD":
+                fpga_raw_path = "data/fpga_raw.bin"  # Not sure where to define this because we don't want to overwrite.
+                with open(fpga_raw_path, "wb") as file:
                     file.write(buf)
                 self.terminate.value = True
             dat = BitArray(buf)
@@ -434,7 +435,7 @@ def _format_frame(
                         f"frame: {header_data.frame_num}, bits lost: {nbit_lost}"
                     )
 
-    def init_video(self, path: Path, fourcc: str = 'Y800', **kwargs) -> cv2.VideoWriter:
+    def init_video(self, path: Path, fourcc: str = "Y800", **kwargs) -> cv2.VideoWriter:
         """
         Create a parameterized video writer
 
@@ -452,13 +453,12 @@ def init_video(self, path: Path, fourcc: str = 'Y800', **kwargs) -> cv2.VideoWri
         out = cv2.VideoWriter(str(path), fourcc, frame_rate, frame_size, **kwargs)
         return out
 
-
     def capture(
         self,
         source: Literal["uart", "fpga"],
         read_length: Optional[int] = None,
         video: Optional[Path] = None,
-        video_kwargs: Optional[dict] = None
+        video_kwargs: Optional[dict] = None,
     ):
         """
         Entry point to start frame capture.
@@ -499,8 +499,8 @@ def capture(
                 target=self._uart_recv,
                 args=(
                     serial_buffer_queue,
-                    self.config['port'],
-                    self.config['baudrate'],
+                    self.config["port"],
+                    self.config["baudrate"],
                 ),
             )
         elif source == "fpga":
@@ -524,7 +524,6 @@ def capture(
         else:
             writer = None
 
-
         def check_termination_flag(termination_flag):
             input("Press enter to exit the process.")
             termination_flag.value = True
@@ -543,23 +542,27 @@ def check_termination_flag(termination_flag):
                 imagearray,
             ),
         )
-        '''
+        """
         p_terminate = multiprocessing.Process(
             target=check_termination_flag, args=(self.terminate,))
-        '''
+        """
         p_recv.start()
         p_buffer_to_frame.start()
         p_format_frame.start()
-        #p_terminate.start()
+        # p_terminate.start()
         try:
             while not self.terminate.value:
                 if imagearray.qsize() > 0:
                     imagearray_plot = imagearray.get()
-                    image = imagearray_plot.reshape(self.config.frame_width, self.config.frame_height)
+                    image = imagearray_plot.reshape(
+                        self.config.frame_width, self.config.frame_height
+                    )
                     if self.config.show_video is True:
                         cv2.imshow("image", image)
                     if writer:
-                        picture = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)  # If your image is grayscale
+                        picture = cv2.cvtColor(
+                            image, cv2.COLOR_GRAY2BGR
+                        )  # If your image is grayscale
                         writer.write(picture)
                 if cv2.waitKey(1) == 27 and self.config.show_video is True:
                     cv2.destroyAllWindows()
@@ -618,7 +621,7 @@ def main():
         except (ValueError, IndexError) as e:
             print(e)
             sys.exit(1)
-        #daq_inst.capture(source="uart", comport=comport, baudrate=baudrate)
+        # daq_inst.capture(source="uart", comport=comport, baudrate=baudrate)
         daq_inst.capture(source="uart")
 
     if daqConfig.device == "OK":

miniscope_io/utils.py

@@ -1,10 +1,11 @@
-from typing import Literal, Union, TYPE_CHECKING
-from pathlib import Path
 import hashlib
+from pathlib import Path
+from typing import TYPE_CHECKING, Union
+
 import cv2
 
 if TYPE_CHECKING:
-    import numpy as np
+    pass
 
 
 def hash_file(path: Union[Path, str]) -> str:
@@ -32,8 +33,8 @@ def hash_file(path: Union[Path, str]) -> str:
 
 
 def hash_video(
-        path: Union[Path, str],
-        method: hashlib.algorithms_available = 'blake2s',
+    path: Union[Path, str],
+    method: hashlib.algorithms_available = "blake2s",
 ) -> str:
     """
     Create a hash of a video by digesting the byte string each of its decoded frames.
@@ -58,9 +59,3 @@ def hash_video(
         h.update(frame)
 
     return h.hexdigest()
-
-
-
-
-
-