Skip to content

Commit

Permalink
Fix pulse mypy errors
Browse files Browse the repository at this point in the history
  • Loading branch information
Randl committed Oct 25, 2023
1 parent 594ae55 commit 682bc65
Show file tree
Hide file tree
Showing 26 changed files with 596 additions and 540 deletions.
154 changes: 76 additions & 78 deletions qiskit/pulse/builder.py

Large diffs are not rendered by default.

32 changes: 17 additions & 15 deletions qiskit/pulse/calibration_entries.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,12 @@
# that they have been altered from the originals.

"""Internal format of calibration data in target."""
from __future__ import annotations
import inspect
from abc import ABCMeta, abstractmethod
from collections.abc import Sequence, Callable
from enum import IntEnum
from typing import Callable, List, Union, Optional, Sequence, Any
from typing import Any

from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.schedule import Schedule, ScheduleBlock
Expand Down Expand Up @@ -79,7 +81,7 @@ def get_signature(self) -> inspect.Signature:
pass

@abstractmethod
def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
def get_schedule(self, *args, **kwargs) -> Schedule | ScheduleBlock:
"""Generate schedule from entry definition.
If the pulse program is templated with :class:`.Parameter` objects,
Expand Down Expand Up @@ -114,7 +116,7 @@ class ScheduleDef(CalibrationEntry):
"""

def __init__(self, arguments: Optional[Sequence[str]] = None):
def __init__(self, arguments: Sequence[str] | None = None):
"""Define an empty entry.
Args:
Expand All @@ -129,9 +131,9 @@ def __init__(self, arguments: Optional[Sequence[str]] = None):
arguments = list(arguments)
self._user_arguments = arguments

self._definition = None
self._signature = None
self._user_provided = None
self._definition: Callable | Schedule | None = None
self._signature: inspect.Signature | None = None
self._user_provided: bool | None = None

@property
def user_provided(self) -> bool:
Expand Down Expand Up @@ -168,7 +170,7 @@ def _parse_argument(self):

def define(
self,
definition: Union[Schedule, ScheduleBlock],
definition: Schedule | ScheduleBlock,
user_provided: bool = True,
):
self._definition = definition
Expand All @@ -178,7 +180,7 @@ def define(
def get_signature(self) -> inspect.Signature:
return self._signature

def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
def get_schedule(self, *args, **kwargs) -> Schedule | ScheduleBlock:
if not args and not kwargs:
out = self._definition
else:
Expand Down Expand Up @@ -252,7 +254,7 @@ def define(
def get_signature(self) -> inspect.Signature:
return self._signature

def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
def get_schedule(self, *args, **kwargs) -> Schedule | ScheduleBlock:
try:
# Python function doesn't allow partial bind, but default value can exist.
to_bind = self._signature.bind(*args, **kwargs)
Expand Down Expand Up @@ -294,9 +296,9 @@ class PulseQobjDef(ScheduleDef):

def __init__(
self,
arguments: Optional[Sequence[str]] = None,
converter: Optional[QobjToInstructionConverter] = None,
name: Optional[str] = None,
arguments: Sequence[str] | None = None,
converter: QobjToInstructionConverter | None = None,
name: str | None = None,
):
"""Define an empty entry.
Expand All @@ -309,7 +311,7 @@ def __init__(

self._converter = converter or QobjToInstructionConverter(pulse_library=[])
self._name = name
self._source = None
self._source: list[PulseQobjInstruction] | None = None

def _build_schedule(self):
"""Build pulse schedule from cmd-def sequence."""
Expand All @@ -322,7 +324,7 @@ def _build_schedule(self):

def define(
self,
definition: List[PulseQobjInstruction],
definition: list[PulseQobjInstruction],
user_provided: bool = False,
):
# This doesn't generate signature immediately, because of lazy schedule build.
Expand All @@ -334,7 +336,7 @@ def get_signature(self) -> inspect.Signature:
self._build_schedule()
return super().get_signature()

def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
def get_schedule(self, *args, **kwargs) -> Schedule | ScheduleBlock:
if self._definition is None:
self._build_schedule()
return super().get_schedule(*args, **kwargs)
Expand Down
14 changes: 9 additions & 5 deletions qiskit/pulse/channels.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,11 +49,13 @@
.. autoclass:: Channel
"""
from __future__ import annotations
from abc import ABCMeta
from typing import Any, Set, Union
from typing import Any

import numpy as np

from qiskit.circuit import Parameter
from qiskit.circuit.parameterexpression import ParameterExpression
from qiskit.pulse.exceptions import PulseError

Expand All @@ -75,7 +77,7 @@ class Channel(metaclass=ABCMeta):
for the ``prefix`` class attribute.
"""

prefix = None # type: Optional[str]
prefix: str | None = None
"""A shorthand string prefix for characterizing the channel type."""

# pylint: disable=unused-argument
Expand All @@ -99,7 +101,7 @@ def __init__(self, index: int):
self._hash = hash((self.__class__.__name__, self._index))

@property
def index(self) -> Union[int, ParameterExpression]:
def index(self) -> int | ParameterExpression:
"""Return the index of this channel. The index is a label for a control signal line
typically mapped trivially to a qubit index. For instance, ``DriveChannel(0)`` labels
the signal line driving the qubit labeled with index 0.
Expand All @@ -125,7 +127,7 @@ def _validate_index(self, index: Any) -> None:
raise PulseError("Channel index must be a nonnegative integer")

@property
def parameters(self) -> Set:
def parameters(self) -> set[Parameter]:
"""Parameters which determine the channel index."""
if isinstance(self.index, ParameterExpression):
return self.index.parameters
Expand All @@ -143,7 +145,7 @@ def name(self) -> str:
def __repr__(self):
return f"{self.__class__.__name__}({self._index})"

def __eq__(self, other: "Channel") -> bool:
def __eq__(self, other: object) -> bool:
"""Return True iff self and other are equal, specifically, iff they have the same type
and the same index.
Expand All @@ -153,6 +155,8 @@ def __eq__(self, other: "Channel") -> bool:
Returns:
True iff equal.
"""
if not isinstance(other, Channel):
return NotImplemented
return type(self) is type(other) and self._index == other._index

def __hash__(self):
Expand Down
37 changes: 21 additions & 16 deletions qiskit/pulse/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,14 @@
"""
Configurations for pulse experiments.
"""
from typing import Dict, Union, Tuple, Optional
from __future__ import annotations
import numpy as np

from .channels import PulseChannel, DriveChannel, MeasureChannel
from .channels import DriveChannel, MeasureChannel
from .exceptions import PulseError


def _assert_nested_dict_equal(a, b):
def _assert_nested_dict_equal(a: dict, b: dict):
if len(a) != len(b):
return False
for key in a:
Expand All @@ -44,7 +44,7 @@ class Kernel:
into IQ points.
"""

def __init__(self, name: Optional[str] = None, **params):
def __init__(self, name: str | None = None, **params):
"""Create new kernel.
Args:
Expand Down Expand Up @@ -72,7 +72,7 @@ class Discriminator:
into 0/1 state results.
"""

def __init__(self, name: Optional[str] = None, **params):
def __init__(self, name: str | None = None, **params):
"""Create new discriminator.
Args:
Expand Down Expand Up @@ -148,8 +148,8 @@ class LoConfig:

def __init__(
self,
channel_los: Optional[Dict[PulseChannel, float]] = None,
lo_ranges: Optional[Dict[PulseChannel, Union[LoRange, Tuple[int]]]] = None,
channel_los: dict[DriveChannel | MeasureChannel, float] | None = None,
lo_ranges: dict[DriveChannel | MeasureChannel, LoRange | tuple[int, int]] | None = None,
):
"""Lo channel configuration data structure.
Expand All @@ -161,9 +161,9 @@ def __init__(
PulseError: If channel is not configurable or set lo is out of range.
"""
self._q_lo_freq = {}
self._m_lo_freq = {}
self._lo_ranges = {}
self._q_lo_freq: dict[DriveChannel, float] = {}
self._m_lo_freq: dict[MeasureChannel, float] = {}
self._lo_ranges: dict[DriveChannel | MeasureChannel, LoRange] = {}

lo_ranges = lo_ranges if lo_ranges else {}
for channel, freq in lo_ranges.items():
Expand All @@ -173,7 +173,7 @@ def __init__(
for channel, freq in channel_los.items():
self.add_lo(channel, freq)

def add_lo(self, channel: Union[DriveChannel, MeasureChannel], freq: float):
def add_lo(self, channel: DriveChannel | MeasureChannel, freq: float):
"""Add a lo mapping for a channel."""
if isinstance(channel, DriveChannel):
# add qubit_lo_freq
Expand All @@ -186,7 +186,9 @@ def add_lo(self, channel: Union[DriveChannel, MeasureChannel], freq: float):
else:
raise PulseError("Specified channel %s cannot be configured." % channel.name)

def add_lo_range(self, channel: DriveChannel, lo_range: Union[LoRange, Tuple[int]]):
def add_lo_range(
self, channel: DriveChannel | MeasureChannel, lo_range: LoRange | tuple[int, int]
):
"""Add lo range to configuration.
Args:
Expand All @@ -198,22 +200,25 @@ def add_lo_range(self, channel: DriveChannel, lo_range: Union[LoRange, Tuple[int
lo_range = LoRange(*lo_range)
self._lo_ranges[channel] = lo_range

def check_lo(self, channel: Union[DriveChannel, MeasureChannel], freq: float) -> bool:
def check_lo(self, channel: DriveChannel | MeasureChannel, freq: float) -> bool:
"""Check that lo is valid for channel.
Args:
channel: Channel to validate lo for
freq: lo frequency
Raises:
PulseError: If freq is outside of channels range
Returns:
True if lo is valid for channel
"""
lo_ranges = self._lo_ranges
if channel in lo_ranges:
lo_range = lo_ranges[channel]
if not lo_range.includes(freq):
raise PulseError(f"Specified LO freq {freq:f} is out of range {lo_range}")
return True

def channel_lo(self, channel: Union[DriveChannel, MeasureChannel]) -> float:
def channel_lo(self, channel: DriveChannel | MeasureChannel) -> float:
"""Return channel lo.
Args:
Expand All @@ -234,11 +239,11 @@ def channel_lo(self, channel: Union[DriveChannel, MeasureChannel]) -> float:
raise PulseError("Channel %s is not configured" % channel)

@property
def qubit_los(self) -> Dict:
def qubit_los(self) -> dict[DriveChannel, float]:
"""Returns dictionary mapping qubit channels (DriveChannel) to los."""
return self._q_lo_freq

@property
def meas_los(self) -> Dict:
def meas_los(self) -> dict[MeasureChannel, float]:
"""Returns dictionary mapping measure channels (MeasureChannel) to los."""
return self._m_lo_freq
35 changes: 21 additions & 14 deletions qiskit/pulse/filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,11 @@
# that they have been altered from the originals.

"""A collection of functions that filter instructions in a pulse program."""

from __future__ import annotations
import abc
from functools import singledispatch
from typing import Callable, List, Union, Iterable, Optional, Tuple, Any
from collections.abc import Iterable
from typing import Callable, Any, List

import numpy as np

Expand All @@ -26,7 +27,10 @@

@singledispatch
def filter_instructions(
sched, filters: List[Callable], negate: bool = False, recurse_subroutines: bool = True
sched,
filters: List[Callable[..., bool]],
negate: bool = False,
recurse_subroutines: bool = True,
):
"""A catch-TypeError function which will only get called if none of the other decorated
functions, namely handle_schedule() and handle_scheduleblock(), handle the type passed.
Expand All @@ -38,7 +42,10 @@ def filter_instructions(

@filter_instructions.register
def handle_schedule(
sched: Schedule, filters: List[Callable], negate: bool = False, recurse_subroutines: bool = True
sched: Schedule,
filters: List[Callable[..., bool]],
negate: bool = False,
recurse_subroutines: bool = True,
) -> Schedule:
"""A filtering function that takes a schedule and returns a schedule consisting of
filtered instructions.
Expand Down Expand Up @@ -79,7 +86,7 @@ def handle_schedule(
@filter_instructions.register
def handle_scheduleblock(
sched_blk: ScheduleBlock,
filters: List[Callable],
filters: List[Callable[..., bool]],
negate: bool = False,
recurse_subroutines: bool = True,
) -> ScheduleBlock:
Expand Down Expand Up @@ -129,11 +136,11 @@ def apply_filters_to_insts_in_scheblk(blk: ScheduleBlock) -> ScheduleBlock:


def composite_filter(
channels: Optional[Union[Iterable[Channel], Channel]] = None,
instruction_types: Optional[Union[Iterable[abc.ABCMeta], abc.ABCMeta]] = None,
time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
intervals: Optional[Iterable[Interval]] = None,
) -> List[Callable]:
channels: Iterable[Channel] | Channel | None = None,
instruction_types: Iterable[abc.ABCMeta] | abc.ABCMeta | None = None,
time_ranges: Iterable[tuple[int, int]] | None = None,
intervals: Iterable[Interval] | None = None,
) -> list[Callable]:
"""A helper function to generate a list of filter functions based on
typical elements to be filtered.
Expand Down Expand Up @@ -163,7 +170,7 @@ def composite_filter(
return filters


def with_channels(channels: Union[Iterable[Channel], Channel]) -> Callable:
def with_channels(channels: Iterable[Channel] | Channel) -> Callable:
"""Channel filter generator.
Args:
Expand Down Expand Up @@ -210,7 +217,7 @@ def handle_instruction(inst: Instruction) -> bool:
return channel_filter


def with_instruction_types(types: Union[Iterable[abc.ABCMeta], abc.ABCMeta]) -> Callable:
def with_instruction_types(types: Iterable[abc.ABCMeta] | abc.ABCMeta) -> Callable:
"""Instruction type filter generator.
Args:
Expand Down Expand Up @@ -257,7 +264,7 @@ def handle_instruction(inst: Instruction) -> bool:
return instruction_filter


def with_intervals(ranges: Union[Iterable[Interval], Interval]) -> Callable:
def with_intervals(ranges: Iterable[Interval] | Interval) -> Callable:
"""Interval filter generator.
Args:
Expand Down Expand Up @@ -286,7 +293,7 @@ def interval_filter(time_inst) -> bool:
return interval_filter


def _if_scalar_cast_to_list(to_list: Any) -> List[Any]:
def _if_scalar_cast_to_list(to_list: Any) -> list[Any]:
"""A helper function to create python list of input arguments.
Args:
Expand Down
Loading

0 comments on commit 682bc65

Please sign in to comment.