Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[QMI-130] Fixing the Mypy check issues as described in the ticket #133

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 24 additions & 18 deletions qmi/core/udp_responder_packets.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import enum
import ctypes
import logging
from typing import ClassVar

from qmi.core.exceptions import QMI_RuntimeException

Expand All @@ -18,7 +19,7 @@ class QMI_LittleEndianStructure(ctypes.LittleEndianStructure):

def __repr__(self) -> str:
fstr_list = []
for (fname, ftype) in self._fields_:
for (fname, ftype, *_) in self._fields_:
value = getattr(self, fname)
if isinstance(value, (QMI_LittleEndianStructure, int, float, bytes)):
vstr = repr(value)
Expand Down Expand Up @@ -71,11 +72,12 @@ class QMI_UdpResponderContextInfoRequestPacket(QMI_UdpResponderPacketHeader):
] # No extra fields.

@staticmethod
def create(pkt_id: int,
pkt_timestamp: float,
workgroup_name_filter: bytes,
context_name_filter: bytes
) -> 'QMI_UdpResponderContextInfoRequestPacket':
def create(
pkt_id: int,
pkt_timestamp: float,
workgroup_name_filter: bytes,
context_name_filter: bytes
) -> 'QMI_UdpResponderContextInfoRequestPacket':
return QMI_UdpResponderContextInfoRequestPacket(
MAGIC,
QMI_UdpResponderMessageTypeTag.CONTEXT_INFO_REQUEST.value,
Expand All @@ -95,15 +97,16 @@ class QMI_UdpResponderContextInfoResponsePacket(QMI_UdpResponderPacketHeader):
]

@staticmethod
def create(pkt_id: int,
pkt_timestamp: float,
request_pkt_id: int,
request_pkt_timestamp: float,
context_pid: int,
context_name: bytes,
workgroup_name: bytes,
context_port: int
) -> 'QMI_UdpResponderContextInfoResponsePacket':
def create(
pkt_id: int,
pkt_timestamp: float,
request_pkt_id: int,
request_pkt_timestamp: float,
context_pid: int,
context_name: bytes,
workgroup_name: bytes,
context_port: int
) -> 'QMI_UdpResponderContextInfoResponsePacket':
context_descriptor = QMI_UdpResponderContextDescriptor(context_pid, context_name, workgroup_name, context_port)
return QMI_UdpResponderContextInfoResponsePacket(
MAGIC, QMI_UdpResponderMessageTypeTag.CONTEXT_INFO_RESPONSE.value,
Expand All @@ -117,7 +120,7 @@ def create(pkt_id: int,

class QMI_UdpResponderKillRequestPacket(QMI_UdpResponderPacketHeader):
_pack_ = 1
_fields_: list = [] # No extra fields.
_fields_: ClassVar[list] = [] # No extra fields.

@staticmethod
def create(pkt_id: int, pkt_timestamp: float) -> 'QMI_UdpResponderKillRequestPacket':
Expand Down Expand Up @@ -157,7 +160,10 @@ def unpack_qmi_udp_packet(packet: bytes) -> QMI_UdpResponderPacketHeader:
expected_packet_size = ctypes.sizeof(packet_type)

if packet_size != expected_packet_size:
raise QMI_RuntimeException("Bad UDP packet: unexpected size (tag = {}, actual = {}, expected = {})".format(
packet_type_tag, packet_size, expected_packet_size))
raise QMI_RuntimeException(
"Bad UDP packet: unexpected size (tag = {}, actual = {}, expected = {})".format(
packet_type_tag, packet_size, expected_packet_size
)
)

return packet_type.from_buffer_copy(packet)
12 changes: 6 additions & 6 deletions qmi/instruments/picoquant/support/_decoders.py
Original file line number Diff line number Diff line change
Expand Up @@ -119,8 +119,8 @@ def process_data(self, fifo_data: np.ndarray) -> np.ndarray:

# Copy the non-overflow records to a result array.
events = np.empty(len(event_types), dtype=EventDataType)
events["type"] = event_types
events["timestamp"] = event_timestamps
events["type"] = event_types # type: ignore[call-overload]
events["timestamp"] = event_timestamps # type: ignore[call-overload]

return events

Expand Down Expand Up @@ -191,10 +191,10 @@ def process_data(self, fifo_data: np.ndarray) -> np.ndarray:
num_data_events = len(event_types)
num_events = num_data_events + len(sync_timestamps)
events = np.empty(num_events, dtype=EventDataType)
events[:num_data_events]["type"] = event_types
events[:num_data_events]["timestamp"] = event_timestamps
events[num_data_events:]["type"] = 64
events[num_data_events:]["timestamp"] = sync_timestamps
events[:num_data_events]["type"] = event_types # type: ignore[call-overload]
events[:num_data_events]["timestamp"] = event_timestamps # type: ignore[call-overload]
events[num_data_events:]["type"] = 64 # type: ignore[call-overload]
events[num_data_events:]["timestamp"] = sync_timestamps # type: ignore[call-overload]
# Events need to be sorted by timestamp, and by type number so that sync# 64 is before channel#.
# NOTE: Since numpy 2.0 overflow of values is not allowed anymore. Therefore, for the means of the lexical sort
# where we use the trick of setting the event types as their negatives to get the right order, now must be
Expand Down
18 changes: 9 additions & 9 deletions qmi/instruments/thorlabs/apt_packets.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Module containing the packets for the APT protocol."""

from typing import List, Tuple
from typing import ClassVar
from qmi.instruments.thorlabs.apt_protocol import (
AptMessage,
AptMessageId,
Expand Down Expand Up @@ -28,7 +28,7 @@ class HW_GET_INFO(AptMessage):
"""

MESSAGE_ID = AptMessageId.HW_GET_INFO.value
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("serial_number", apt_long),
("model_number", apt_char * 8),
("type", apt_word),
Expand Down Expand Up @@ -58,7 +58,7 @@ class MOD_GET_CHANENABLESTATE(AptMessage):

MESSAGE_ID = AptMessageId.MOD_GET_CHANENABLESTATE.value
HEADER_ONLY = True
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("message_id", apt_word),
("chan_ident", apt_byte),
("enable_state", apt_byte),
Expand All @@ -82,7 +82,7 @@ class MOT_MOVE_HOMED(AptMessage):

MESSAGE_ID = AptMessageId.MOT_MOVE_HOMED.value
HEADER_ONLY = True
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("message_id", apt_word),
("chan_ident", apt_byte),
("param2", apt_byte),
Expand All @@ -101,7 +101,7 @@ class MOT_MOVE_ABSOLUTE(AptMessage):
"""

MESSAGE_ID = AptMessageId.MOT_MOVE_ABSOLUTE.value
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("chan_ident", apt_word),
("absolute_distance", apt_long),
]
Expand All @@ -122,7 +122,7 @@ class MOT_MOVE_COMPLETED(AptMessage):

MESSAGE_ID = AptMessageId.MOT_MOVE_COMPLETED.value
HEADER_ONLY = True
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("message_id", apt_word),
("chan_ident", apt_byte),
("param2", apt_byte),
Expand All @@ -145,7 +145,7 @@ class MOT_GET_USTATUSUPDATE(AptMessage):
"""

MESSAGE_ID = AptMessageId.MOT_GET_USTATUSUPDATE.value
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("chan_ident", apt_word),
("position", apt_long),
("velocity", apt_word),
Expand All @@ -164,7 +164,7 @@ class MOT_SET_EEPROMPARAMS(AptMessage):
"""

MESSAGE_ID = AptMessageId.MOT_SET_EEPROMPARAMS.value
_fields_: List[Tuple[str, type]] = [("chan_ident", apt_word), ("msg_id", apt_word)]
_fields_: ClassVar[list[tuple[str, type]]] = [("chan_ident", apt_word), ("msg_id", apt_word)]


class POL_GET_SET_PARAMS(AptMessage):
Expand All @@ -181,7 +181,7 @@ class POL_GET_SET_PARAMS(AptMessage):
"""

MESSAGE_ID = AptMessageId.POL_GET_PARAMS.value
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("not_used", apt_word),
("velocity", apt_word),
("home_position", apt_word),
Expand Down
10 changes: 5 additions & 5 deletions qmi/instruments/thorlabs/apt_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
sizeof,
)
from enum import Enum
from typing import List, Optional, Tuple, Type, TypeVar
from typing import ClassVar, Type, TypeVar

from qmi.core.transport import QMI_Transport
from qmi.core.exceptions import QMI_InstrumentException
Expand Down Expand Up @@ -111,7 +111,7 @@ class AptMessageHeaderWithParams(LittleEndianStructure):
"""

_pack_ = True
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("message_id", c_uint16),
("param1", c_uint8),
("param2", c_uint8),
Expand All @@ -126,7 +126,7 @@ class AptMessageHeaderForData(LittleEndianStructure):
"""

_pack_ = True
_fields_: List[Tuple[str, type]] = [
_fields_: ClassVar[list[tuple[str, type]]] = [
("message_id", c_uint16),
("data_length", c_uint16),
("dest", c_uint8),
Expand Down Expand Up @@ -159,7 +159,7 @@ def __init__(
transport: QMI_Transport,
apt_device_address: int = 0x50,
host_address: int = 0x01,
default_timeout: Optional[float] = None,
default_timeout: float | None = None,
):
"""Initialize the Thorlabs APT protocol handler.

Expand Down Expand Up @@ -215,7 +215,7 @@ def write_data_command(self, message_id: int, data: T) -> None:
# Send the header and data packet.
self._transport.write(bytearray(msg) + bytearray(data))

def ask(self, data_type: Type[T], timeout: Optional[float] = None) -> T:
def ask(self, data_type: Type[T], timeout: float | None = None) -> T:
"""
Ask for a response.

Expand Down
Loading
Loading