Skip to content

Commit

Permalink
[pre-commit.ci] auto fixes from pre-commit.com hooks
Browse files Browse the repository at this point in the history
for more information, see https://pre-commit.ci
  • Loading branch information
pre-commit-ci[bot] committed Sep 30, 2024
1 parent 86643c7 commit d602e22
Show file tree
Hide file tree
Showing 16 changed files with 59 additions and 70 deletions.
6 changes: 2 additions & 4 deletions pyseq2/base/instruments.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,7 @@ class UsesSerial(metaclass=ABCMeta):
com: COM

@abstractmethod
async def initialize(self) -> None:
...
async def initialize(self) -> None: ...


class Movable(metaclass=ABCMeta):
Expand All @@ -28,8 +27,7 @@ async def move(self, pos: int) -> None:

@property
@abstractmethod
async def pos(self) -> int:
...
async def pos(self) -> int: ...

def convert(self, p: Annotated[float, "mm"]) -> int:
return int(p * 1000 * self.STEPS_PER_UM)
Expand Down
6 changes: 2 additions & 4 deletions pyseq2/com/async_com.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,12 +212,10 @@ async def _read_forever(self) -> NoReturn:
logger.critical(f"{self.name}{type(e).__name__}: {e}")

@overload
async def send(self, cmd: str) -> None:
...
async def send(self, cmd: str) -> None: ...

@overload
async def send(self, cmd: CmdParse[Any, T]) -> T:
...
async def send(self, cmd: CmdParse[Any, T]) -> T: ...

async def send(self, cmd: str | CmdParse[Any, T]) -> None | T:
"""Sends a command to an instrument.
Expand Down
6 changes: 2 additions & 4 deletions pyseq2/experiment/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,11 @@ class AbstractCommand(metaclass=ABCMeta):
op: str

@abstractmethod
async def run(self, fcs: FlowCells, i: bool, imager: Imager) -> Any:
...
async def run(self, fcs: FlowCells, i: bool, imager: Imager) -> Any: ...

@classmethod
@abstractmethod
def default(cls: type[T]) -> T:
...
def default(cls: type[T]) -> T: ...


async def pump_prime(fcs: FlowCells, i: bool, cmd: Pump | Prime):
Expand Down
2 changes: 1 addition & 1 deletion pyseq2/experiment/reagent.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#%%
# %%
from __future__ import annotations

from collections import defaultdict
Expand Down
3 changes: 1 addition & 2 deletions pyseq2/fakes/fake_handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,7 @@


class Fakes(Protocol):
def __call__(self, s: str) -> str:
...
def __call__(self, s: str) -> str: ...


class FakeX:
Expand Down
2 changes: 1 addition & 1 deletion pyseq2/fluidics/arm9chem.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ async def set_vacuum(self, onoff: bool) -> None:
await self.com.send(ARM9Cmd.SET_VACUUM(onoff))

@asynccontextmanager
async def shutoff_valve(self) -> AsyncGenerator[None, None]:
async def shutoff_valve(self) -> AsyncGenerator[None]:
try:
await self.com.send(ARM9Cmd.SET_SHUTOFF_VALVE(1))
yield
Expand Down
2 changes: 1 addition & 1 deletion pyseq2/fluidics/pump.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ async def _pump(
v_pull: Sps = 400,
v_push: Sps = 6400,
reverse: bool = False,
) -> AsyncGenerator[None, None]:
) -> AsyncGenerator[None]:
try:
logger.info(f"Pump {self.name}:{' Reverse' if reverse else ''} Pulling. ")
await self._pushpull("pull", vol, speed=v_pull, reverse=reverse)
Expand Down
8 changes: 3 additions & 5 deletions pyseq2/fluidics/valve.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,7 @@ class ValveCmd:
# fmt: on


class ValveError(Exception):
...
class ValveError(Exception): ...


class _Valve(Movable, UsesSerial):
Expand All @@ -53,8 +52,7 @@ def __init__(self, name: ValveName, n_ports: Literal[10, 24], com: COM) -> None:
self.name = name
self.t_lastcmd = 0.0

async def initialize(self) -> None:
...
async def initialize(self) -> None: ...

@property
async def pos(self) -> int:
Expand Down Expand Up @@ -156,7 +154,7 @@ async def move(self, pos: int) -> None:
raise NotImplementedError("Use the async context manager move_port instead.")

@asynccontextmanager
async def move_port(self, pos: int) -> AsyncGenerator[None, None]:
async def move_port(self, pos: int) -> AsyncGenerator[None]:
try:
await self._move(pos)
yield
Expand Down
15 changes: 6 additions & 9 deletions pyseq2/imaging/camera/dcam.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,8 +114,7 @@ def init_properties(handle: c_void_p) -> DCAMDict:
else:
return DCAMDict.from_dcam(handle)

def initialize(self) -> None:
...
def initialize(self) -> None: ...

@property
def capture_mode(self) -> DCAM_CAPTURE_MODE:
Expand All @@ -127,7 +126,7 @@ def capture_mode(self, m: DCAM_CAPTURE_MODE) -> None:
self._capture_mode = m

@contextmanager
def capture(self) -> Generator[None, None, None]:
def capture(self) -> Generator[None]:
API.dcam_capture(self.handle)
logger.info(f"Cam {self.id_}: Started capturing.")
try:
Expand All @@ -146,7 +145,7 @@ def status(self) -> Status:
raise DCAMException(f"Invalid status. Got {s.value}.")

@contextmanager
def attach(self, n_bundles: int, dim: tuple[int, int]) -> Generator[UInt16Array, None, None]:
def attach(self, n_bundles: int, dim: tuple[int, int]) -> Generator[UInt16Array]:
"""Generates a numpy array and "attach" it to the camera.
Aka. Tells the camera to write captured bundles here.
Expand Down Expand Up @@ -255,20 +254,18 @@ def n_frames_taken(self, cam: Cam) -> int:
@contextmanager
def _attach(
self, n_bundles: int, dim: tuple[int, int], cam: Literal[0, 1] = ...
) -> Generator[UInt16Array, None, None]:
...
) -> Generator[UInt16Array]: ...

@overload
@contextmanager
def _attach(
self, n_bundles: int, dim: tuple[int, int], cam: Literal[2] = ...
) -> Generator[tuple[UInt16Array, UInt16Array], None, None]:
...
) -> Generator[tuple[UInt16Array, UInt16Array]]: ...

@contextmanager
def _attach(
self, n_bundles: int, dim: tuple[int, int], cam: Cam = 2
) -> Generator[UInt16Array, None, None] | Generator[tuple[UInt16Array, UInt16Array], None, None]:
) -> Generator[UInt16Array] | Generator[tuple[UInt16Array, UInt16Array]]:
if cam == 2:
with self[0].attach(n_bundles, dim) as buf1, self[1].attach(n_bundles, dim) as buf2:
logger.debug(f"Allocated memory for {n_bundles} bundles.")
Expand Down
6 changes: 2 additions & 4 deletions pyseq2/imaging/camera/dcam_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,12 +67,10 @@ def __getattribute__(self, __name: str) -> Any:
return f


class DCAMException(Exception):
...
class DCAMException(Exception): ...


class DCAMReturnedZero(DCAMException):
...
class DCAMReturnedZero(DCAMException): ...


class DCAM_CAPTURE_MODE(IntEnum):
Expand Down
19 changes: 19 additions & 0 deletions pyseq2/imaging/camera/dcam_api.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -44,40 +44,54 @@ class DCAMAPI(WinDLL):
def dcam_queryupdate(self, h: Handle, pFlag: pointer[c_int32], reserved: pointer[c_int32]) -> bool: ...
def dcam_getbinning(self, h: Handle, pBinning: pointer[c_int32]) -> bool:
"""This will return the current binning mode of the camera."""

def dcam_getexposuretime(self, h: Handle, pSec: pointer[c_double]) -> bool:
"""This returns the value of the current exposure time in seconds."""

def dcam_gettriggermode(self, h: Handle, pMode: pointer[c_int32]) -> bool:
"""This returns the value of the current trigger mode. The value returned will be numeric and it may be necessary to use the global variables to process this data."""

def dcam_gettriggerpolarity(self, h: Handle, pPolarity: pointer[c_int32]) -> bool:
"""This returns the value of the current polarity. This function will fail if the camera does not support external trigger."""

def dcam_setbinning(self, h: Handle, binning: c_int32) -> bool:
"""This function will adjust the binning of the camera. Adjusting the binning will also change the output data size, but the pixel depth will remain the same."""

def dcam_setexposuretime(self, h: Handle, sec: c_double) -> bool:
"""This allows you to modify the exposure time in seconds of the camera. Depending on the camera, the exposure time that you specify may not be the same exposure time actually set. If necessary, check the actual time set using DCAM_GETEXPOSURETIME."""

def dcam_settriggermode(self, h: Handle, mode: c_int32) -> bool:
"""This function allows you to switch the current trigger mode. Create a constant or control from the Mode terminal to get a list of trigger mode IDs. However, although they are listed, it may or may not be a valid ID for your camera."""

def dcam_settriggerpolarity(self, h: Handle, polarity: c_int32) -> bool:
"""This will select the polarity of the trigger if the camera has been set to external trigger mode. This function will fail if the camera does not support external trigger. Create a constant or control from the Polarity terminal to get a list of valid IDs."""
# /*** --- capturing --- ***/
def dcam_precapture(self, h: Handle, mode: DCAM_CAPTURE_MODE) -> bool:
"""Sets the capture mode and prepares resources for capturing. This also changes the camera status to STABLE state. Create a constant or control from the Capture Mode terminal to get a list of valid IDs."""

def dcam_getdatarange(
self,
h: Handle,
pMax: pointer[c_int32],
pMin: pointer[c_int32] = ...,
) -> bool:
"""This function will return the minimum and maximum values possible for the data pixels."""

def dcam_getdataframebytes(self, h: Handle, pSize: pointer[c_int32]) -> bool:
"""This determines the amount of bytes required for a single frame of data. This is useful when using DCAM_ATTACHBUFFER."""

def dcam_allocframe(self, h: Handle, framecount: c_int32) -> bool:
"""Allocates the proper amount of memory for a data buffer depending on the number of frames requested and changes the camera status from STABLE to READY. Capturing does not start at this time. This function will fail if called while camera is not in STABLE state."""

def dcam_getframecount(self, h: Handle, pFrame: pointer[c_int32]) -> bool:
"""Returns the number of frames allocated in memory by either DCAM_ALLOCFRAME or by DCAM_ATTACHBUFFER. This function will fail is no frames have been allocated."""

def dcam_capture(self, h: Handle) -> bool:
"""This starts capturing images. Camera status must be READY before using this function. If the camera is in SEQUENCE mode, the camera will capture images repeatedly. If the camera is in SNAP mode, the camera will capture only the number of images for frames that were allocated."""

def dcam_idle(self, h: Handle) -> bool:
"""This function will stop the capturing of images. If the camera is in BUSY state, it will set it to READY state."""

def dcam_wait(
self,
h: Handle,
Expand All @@ -87,6 +101,7 @@ class DCAMAPI(WinDLL):
) -> bool: ...
def dcam_getstatus(self, h: Handle, pStatus: pointer[c_int32]) -> bool:
"""This returns the state of the current camera operation."""

def dcam_gettransferinfo(
self,
h: Handle,
Expand All @@ -98,6 +113,7 @@ class DCAMAPI(WinDLL):
# /*** --- user memory support --- ***/
def dcam_attachbuffer(self, h: Handle, frames: Array[c_void_p], size: c_uint32) -> bool:
"""This allows the user to attach his/her own data buffer instead of using DCAM_ALLOCFRAME and using the DCAM buffer. This function accepts an array of pointers to image buffers. If used, this will take the place of DCAM_ALLOCFRAME and set the camera from STABLE to READY state."""

def dcam_releasebuffer(self, h: Handle) -> bool:
"""This function works similar to DCAM_FREEFRAME except that this is used when using DCAM_ATTACHBUFFER. This will set the camera from READY to STABLE state."""
# /*** --- data transfer --- ***/
Expand All @@ -109,6 +125,7 @@ class DCAMAPI(WinDLL):
frame: c_int32,
) -> bool:
"""This locks a frame in the data buffer that will be accessed by the user. When a frame is locked, the capture thread will not write to it. Calling this function will also unlock any existing locked frames. If you set the frame input as -1, you will lock the most recently received frame. Put the frame data address into pTop."""

def dcam_lockbits(
self,
h: Handle,
Expand All @@ -117,8 +134,10 @@ class DCAMAPI(WinDLL):
frame: c_int32,
) -> bool:
"""This works similar to DCAM_LOCKDATA except that the buffer data has been altered for display. For example, 16 bit images are converted to 8 bit images and flipped for easy use in Windows API functions. If you set the frame input as -1, you will lock the most recently received frame."""

def dcam_unlockdata(self, h: Handle) -> bool:
"""If a framed has been locked by DCAM_LOCKDATA, this function will unlock it."""

def dcam_unlockbits(self, h: Handle) -> bool:
"""If a framed has been locked by DCAM_LOCKBITS, this function will unlock it."""
# /*** --- LUT --- ***/
Expand Down
3 changes: 1 addition & 2 deletions pyseq2/imaging/laser.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,7 @@
mW = Annotated[int, "mW"]


class LaserException(Exception):
...
class LaserException(Exception): ...


def v_get_status(resp: str) -> bool:
Expand Down
24 changes: 8 additions & 16 deletions pyseq2/utils/coords.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,23 +11,19 @@


@overload
def raw_to_mm(flowcell: bool, *, x: int, y: None = None) -> float:
...
def raw_to_mm(flowcell: bool, *, x: int, y: None = None) -> float: ...


@overload
def raw_to_mm(flowcell: bool, *, y: int, x: None = None) -> float:
...
def raw_to_mm(flowcell: bool, *, y: int, x: None = None) -> float: ...


@overload
def raw_to_mm(flowcell: bool, *, x: int, y: int) -> tuple[float, float]:
...
def raw_to_mm(flowcell: bool, *, x: int, y: int) -> tuple[float, float]: ...


@overload
def raw_to_mm(flowcell: bool, *, x: None = None, y: None = None) -> None:
...
def raw_to_mm(flowcell: bool, *, x: None = None, y: None = None) -> None: ...


def raw_to_mm(
Expand All @@ -45,23 +41,19 @@ def raw_to_mm(


@overload
def mm_to_raw(flowcell: bool, *, x: float, y: None = None) -> int:
...
def mm_to_raw(flowcell: bool, *, x: float, y: None = None) -> int: ...


@overload
def mm_to_raw(flowcell: bool, *, y: float, x: None = None) -> int:
...
def mm_to_raw(flowcell: bool, *, y: float, x: None = None) -> int: ...


@overload
def mm_to_raw(flowcell: bool, *, x: float, y: float) -> tuple[int, int]:
...
def mm_to_raw(flowcell: bool, *, x: float, y: float) -> tuple[int, int]: ...


@overload
def mm_to_raw(flowcell: bool, *, x: None = None, y: None = None) -> None:
...
def mm_to_raw(flowcell: bool, *, x: None = None, y: None = None) -> None: ...


def mm_to_raw(
Expand Down
2 changes: 1 addition & 1 deletion pyseq2/utils/ports.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#%%
# %%
import asyncio
import logging
from pprint import pprint
Expand Down
19 changes: 7 additions & 12 deletions pyseq2/utils/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

T, P = TypeVar("T"), ParamSpec("P")


# https://python-3-patterns-idioms-test.readthedocs.io/en/latest/Metaprogramming.html#intercepting-class-creation
class Singleton(type):
instance = None
Expand All @@ -19,12 +20,10 @@ def __call__(cls, *args: Any, **kwargs: Any) -> Singleton:
return cls.instance


class InvalidResponse(Exception):
...
class InvalidResponse(Exception): ...


class ParamChangeTimeout(Exception):
...
class ParamChangeTimeout(Exception): ...


def ok_if_match(expected: Container[str] | str, exception_on_fail: bool = True) -> Callable[[str], bool]:
Expand Down Expand Up @@ -81,13 +80,11 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:


@overload
def λ_int(λ: Callable[[Any, Any], T]) -> Callable[[int, int], T]:
...
def λ_int(λ: Callable[[Any, Any], T]) -> Callable[[int, int], T]: ...


@overload
def λ_int(λ: Callable[[Any], T]) -> Callable[[int], T]:
...
def λ_int(λ: Callable[[Any], T]) -> Callable[[int], T]: ...


def λ_int(λ: Callable[[Any], T] | Callable[[Any, Any], T]) -> Callable[[int], T] | Callable[[int, int], T]:
Expand All @@ -101,13 +98,11 @@ def inner(*args: int) -> T:


@overload
def λ_float(λ: Callable[[Any, Any], T]) -> Callable[[IntFloat, IntFloat], T]:
...
def λ_float(λ: Callable[[Any, Any], T]) -> Callable[[IntFloat, IntFloat], T]: ...


@overload
def λ_float(λ: Callable[[Any], T]) -> Callable[[IntFloat], T]:
...
def λ_float(λ: Callable[[Any], T]) -> Callable[[IntFloat], T]: ...


def λ_float(
Expand Down
Loading

0 comments on commit d602e22

Please sign in to comment.