diff --git a/.github/workflows/quam_test.yaml b/.github/workflows/quam_test.yaml index ddcea773..f1bdebcf 100644 --- a/.github/workflows/quam_test.yaml +++ b/.github/workflows/quam_test.yaml @@ -7,6 +7,32 @@ on: branches: [ main ] jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: "pip" + cache-dependency-path: pyproject.toml + + - name: Install dev dependencies + run: | + python -m pip install --upgrade pip + pip install .[dev] + + - name: Black formatting check + run: poe check-format + + - name: Flake8 linting + run: poe lint + + - name: Mypy type check + run: poe typecheck + run-tests: uses: ./.github/workflows/run_tests.yaml secrets: inherit diff --git a/CHANGELOG.md b/CHANGELOG.md index c0e318f1..44543b92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ ## [Unreleased] +### Added + +- Added `mypy` type checking and a `poethepoet` task runner to the dev workflow: `poe format` / `poe check-format` (black), `poe lint` (flake8), `poe typecheck` (mypy), `poe test` (pytest), and a combined `poe check` that runs them all. + +### Changed + +- Annotated the codebase and resolved all `mypy` errors, alongside black formatting. These are internal type-correctness and formatting changes only — no public API or runtime behavior changes. + ## [v0.6.0] ### Added diff --git a/pyproject.toml b/pyproject.toml index 8a8ffba5..e923612b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,9 +51,11 @@ dev = [ "black >= 23.7.0", "flake8 >= 5.0.1", "pyproject-flake8 >= 5.0.0", + "mypy >= 1.0.0", "ipykernel >= 6.24.0", "pytest-cov >= 4.1.0", "pytest-mock >= 3.6.1", + "poethepoet >= 0.24.0", "qm-saas", ] docs = ["mkdocstrings[python]>=0.18", "mkdocs-gen-files", "mkdocs-jupyter"] @@ -101,6 +103,9 @@ preview = true [tool.flake8] max-line-length = 88 +# Black handles line length; E501 fires on lines it can't shorten (type: ignore comments, docstrings). +extend-ignore = ["E501"] +per-file-ignores = ["**/__init__.py:F401,F403,F405"] [tool.setuptools] packages = ["quam"] @@ -112,38 +117,28 @@ local_scheme = "no-local-version" [tool.uv] prerelease = "allow" -# Previous quam settings +[tool.poe.tasks.format] +cmd = "black quam/" +help = "Auto-format source files" -### Extra required dependencies -# "jsonschema = ^4.14.0", -# "pydantic = ^1.9.2", -# "rich = ^12.5.1", -# "rich-click = ^1.6.1", -# "pyyaml = ^6.0", +[tool.poe.tasks.check-format] +cmd = "black --check quam/" +help = "Check formatting without modifying files" -### Extra optional dependencies -# flake8-bugbear = "^22.4.25" -# poethepoet = "^0.10.0" -# typing-extensions = "^4.3.0" -# coverage = "^6.4.4" +[tool.poe.tasks.lint] +cmd = "pflake8 quam/" +help = "Run flake8 linting" +[tool.poe.tasks.typecheck] +# --ignore-missing-imports: suppress errors for deps without type stubs (e.g. qm-qua). +# --follow-imports=silent: collect type info from followed imports but suppress errors inside them. +cmd = "mypy quam/ --ignore-missing-imports --follow-imports=silent" +help = "Run mypy type checking" -# [tool.poe.tasks.check-format] -# cmd = "black . --check --line-length 80" -# help = "Format source files according to the style rules" +[tool.poe.tasks.test] +cmd = "pytest" +help = "Run all unit tests" -# [tool.poe.tasks.format] -# cmd = "black . --line-length 80" -# help = "Format source files according to the style rules" - -# [tool.poe.tasks.lint] -# cmd = "flake8 ." -# help = "Check for lint errors" - -# [tool.poe.tasks.test] -# cmd = "pytest" -# help = "Run all unit tests" - -# [tool.poe.tasks.check] -# sequence = ["check-format", "lint", "test"] -# help = "Run format and all checks on the code" +[tool.poe.tasks.check] +sequence = ["check-format", "lint", "typecheck", "test"] +help = "Run all checks (format, lint, types, tests)" diff --git a/quam/components/__init__.py b/quam/components/__init__.py index 2c9a8049..710c2d9a 100644 --- a/quam/components/__init__.py +++ b/quam/components/__init__.py @@ -6,6 +6,7 @@ from .quantum_components import * from . import macro from quam.config import get_quam_config +from . import basic_quam, hardware, channels, octave, quantum_components __all__ = [ *basic_quam.__all__, diff --git a/quam/components/_waveform_tools.py b/quam/components/_waveform_tools.py index 6c370290..a0e5e07d 100644 --- a/quam/components/_waveform_tools.py +++ b/quam/components/_waveform_tools.py @@ -8,7 +8,15 @@ def drag_gaussian_pulse_waveforms( - amplitude, length, sigma, alpha, anharmonicity, detuning=0.0, subtracted=True, sampling_rate=1e9, **kwargs + amplitude, + length, + sigma, + alpha, + anharmonicity, + detuning=0.0, + subtracted=True, + sampling_rate=1e9, + **kwargs, ): if alpha != 0 and anharmonicity == 0: raise ValueError("Cannot create a DRAG pulse with `anharmonicity=0`") @@ -16,13 +24,19 @@ def drag_gaussian_pulse_waveforms( center = (length - 1e9 / sampling_rate) / 2 gauss_wave = amplitude * np.exp(-((t - center) ** 2) / (2 * sigma**2)) gauss_der_wave = ( - amplitude * (-2 * 1e9 * (t - center) / (2 * sigma**2)) * np.exp(-((t - center) ** 2) / (2 * sigma**2)) + amplitude + * (-2 * 1e9 * (t - center) / (2 * sigma**2)) + * np.exp(-((t - center) ** 2) / (2 * sigma**2)) ) if subtracted: gauss_wave = gauss_wave - gauss_wave[-1] z = gauss_wave + 1j * 0 if anharmonicity != detuning: - z += 1j * gauss_der_wave * (alpha / (2 * np.pi * anharmonicity - 2 * np.pi * detuning)) + z += ( + 1j + * gauss_der_wave + * (alpha / (2 * np.pi * anharmonicity - 2 * np.pi * detuning)) + ) elif alpha != 0: raise ValueError( "Cannot create DRAG waveform if anharmonicity == detuning and alpha != 0." @@ -31,16 +45,25 @@ def drag_gaussian_pulse_waveforms( return z.real.tolist(), z.imag.tolist() -def drag_cosine_pulse_waveforms(amplitude, length, alpha, anharmonicity, detuning=0.0, sampling_rate=1e9, **kwargs): +def drag_cosine_pulse_waveforms( + amplitude, length, alpha, anharmonicity, detuning=0.0, sampling_rate=1e9, **kwargs +): if alpha != 0 and anharmonicity == 0: raise ValueError("Cannot create a DRAG pulse with `anharmonicity=0`") t = np.arange(length, step=1e9 / sampling_rate) end_point = length - 1e9 / sampling_rate cos_wave = 0.5 * amplitude * (1 - np.cos(t * 2 * np.pi / end_point)) - sin_wave = 0.5 * amplitude * (2 * np.pi / end_point * 1e9) * np.sin(t * 2 * np.pi / end_point) + sin_wave = ( + 0.5 + * amplitude + * (2 * np.pi / end_point * 1e9) + * np.sin(t * 2 * np.pi / end_point) + ) z = cos_wave + 1j * 0 if anharmonicity != detuning: - z += 1j * sin_wave * (alpha / (2 * np.pi * anharmonicity - 2 * np.pi * detuning)) + z += ( + 1j * sin_wave * (alpha / (2 * np.pi * anharmonicity - 2 * np.pi * detuning)) + ) elif alpha != 0: raise ValueError( "Cannot create DRAG waveform if anharmonicity == detuning and alpha != 0." @@ -49,15 +72,23 @@ def drag_cosine_pulse_waveforms(amplitude, length, alpha, anharmonicity, detunin return z.real.tolist(), z.imag.tolist() -def flattop_gaussian_waveform(amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9): - assert sampling_rate % 1e9 == 0, "The sampling rate must be an integer multiple of 1e9 samples per second." +def flattop_gaussian_waveform( + amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9 +): + assert ( + sampling_rate % 1e9 == 0 + ), "The sampling rate must be an integer multiple of 1e9 samples per second." gauss_wave = amplitude * gaussian( int(np.round(2 * rise_fall_length * sampling_rate / 1e9)), rise_fall_length / 5 * sampling_rate / 1e9, ) rise_part = gauss_wave[: int(rise_fall_length * sampling_rate / 1e9)].tolist() if return_part == "all": - return rise_part + [amplitude] * int(flat_length * sampling_rate / 1e9) + rise_part[::-1] + return ( + rise_part + + [amplitude] * int(flat_length * sampling_rate / 1e9) + + rise_part[::-1] + ) elif return_part == "rise": return rise_part elif return_part == "fall": @@ -65,13 +96,26 @@ def flattop_gaussian_waveform(amplitude, flat_length, rise_fall_length, return_p raise ValueError("'return_part' must be 'all', 'rise', or 'fall'") -def flattop_cosine_waveform(amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9): - assert sampling_rate % 1e9 == 0, "The sampling rate must be an integer multiple of 1e9 samples per second." +def flattop_cosine_waveform( + amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9 +): + assert ( + sampling_rate % 1e9 == 0 + ), "The sampling rate must be an integer multiple of 1e9 samples per second." rise_part = ( - amplitude * 0.5 * (1 - np.cos(np.linspace(0, np.pi, int(rise_fall_length * sampling_rate / 1e9)))) + amplitude + * 0.5 + * ( + 1 + - np.cos(np.linspace(0, np.pi, int(rise_fall_length * sampling_rate / 1e9))) + ) ).tolist() if return_part == "all": - return rise_part + [amplitude] * int(flat_length * sampling_rate / 1e9) + rise_part[::-1] + return ( + rise_part + + [amplitude] * int(flat_length * sampling_rate / 1e9) + + rise_part[::-1] + ) elif return_part == "rise": return rise_part elif return_part == "fall": @@ -79,13 +123,23 @@ def flattop_cosine_waveform(amplitude, flat_length, rise_fall_length, return_par raise ValueError("'return_part' must be 'all', 'rise', or 'fall'") -def flattop_tanh_waveform(amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9): - assert sampling_rate % 1e9 == 0, "The sampling rate must be an integer multiple of 1e9 samples per second." +def flattop_tanh_waveform( + amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9 +): + assert ( + sampling_rate % 1e9 == 0 + ), "The sampling rate must be an integer multiple of 1e9 samples per second." rise_part = ( - amplitude * 0.5 * (1 + np.tanh(np.linspace(-4, 4, int(rise_fall_length * sampling_rate / 1e9)))) + amplitude + * 0.5 + * (1 + np.tanh(np.linspace(-4, 4, int(rise_fall_length * sampling_rate / 1e9)))) ).tolist() if return_part == "all": - return rise_part + [amplitude] * int(flat_length * sampling_rate / 1e9) + rise_part[::-1] + return ( + rise_part + + [amplitude] * int(flat_length * sampling_rate / 1e9) + + rise_part[::-1] + ) elif return_part == "rise": return rise_part elif return_part == "fall": @@ -93,12 +147,22 @@ def flattop_tanh_waveform(amplitude, flat_length, rise_fall_length, return_part= raise ValueError("'return_part' must be 'all', 'rise', or 'fall'") -def flattop_blackman_waveform(amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9): - assert sampling_rate % 1e9 == 0, "The sampling rate must be an integer multiple of 1e9 samples per second." - blackman_wave = amplitude * blackman(2 * int(rise_fall_length * sampling_rate / 1e9)) +def flattop_blackman_waveform( + amplitude, flat_length, rise_fall_length, return_part="all", sampling_rate=1e9 +): + assert ( + sampling_rate % 1e9 == 0 + ), "The sampling rate must be an integer multiple of 1e9 samples per second." + blackman_wave = amplitude * blackman( + 2 * int(rise_fall_length * sampling_rate / 1e9) + ) rise_part = blackman_wave[: int(rise_fall_length * sampling_rate / 1e9)].tolist() if return_part == "all": - return rise_part + [amplitude] * int(flat_length * sampling_rate / 1e9) + rise_part[::-1] + return ( + rise_part + + [amplitude] * int(flat_length * sampling_rate / 1e9) + + rise_part[::-1] + ) elif return_part == "rise": return rise_part elif return_part == "fall": @@ -107,7 +171,9 @@ def flattop_blackman_waveform(amplitude, flat_length, rise_fall_length, return_p def blackman_integral_waveform(pulse_length, v_start, v_end, sampling_rate=1e9): - assert sampling_rate % 1e9 == 0, "The sampling rate must be an integer multiple of 1e9 samples per second." + assert ( + sampling_rate % 1e9 == 0 + ), "The sampling rate must be an integer multiple of 1e9 samples per second." time = np.linspace(0, pulse_length - 1, int(pulse_length * sampling_rate / 1e9)) waveform = v_start + ( time / (pulse_length - 1) @@ -131,15 +197,27 @@ def compress_integration_weights(integration_weights, N=100): integration_weights[idx, 0] = (w1 * t1 + w2 * t2) / (t1 + t2) integration_weights[idx, 1] = t1 + t2 integration_weights = np.delete(integration_weights, idx + 1, 0) - return list(zip(integration_weights.T[0].tolist(), integration_weights.T[1].astype(int).tolist())) + return list( + zip( + integration_weights.T[0].tolist(), + integration_weights.T[1].astype(int).tolist(), + ) + ) def convert_integration_weights(integration_weights, N=100, accuracy=2**-15): - integration_weights = _round_to_fixed_point_accuracy(np.array(integration_weights), accuracy) + integration_weights = _round_to_fixed_point_accuracy( + np.array(integration_weights), accuracy + ) changes_indices = np.where(np.abs(np.diff(integration_weights)) > 0)[0].tolist() prev_index = -1 new_weights = [] for curr_index in changes_indices + [len(integration_weights) - 1]: - new_weights.append((integration_weights[curr_index].tolist(), round(4 * (curr_index - prev_index)))) + new_weights.append( + ( + integration_weights[curr_index].tolist(), + round(4 * (curr_index - prev_index)), + ) + ) prev_index = curr_index return compress_integration_weights(new_weights, N=N) diff --git a/quam/components/channels.py b/quam/components/channels.py index db963d6e..08950c81 100644 --- a/quam/components/channels.py +++ b/quam/components/channels.py @@ -149,23 +149,25 @@ class DigitalOutputChannel(QuamComponent): 136 ns exists by default. buffer (int, optional): Digital pulses played to this element will be convolved with a digital pulse of value 1 with this length [ns]. - shareable (bool, deprecated): If True, the digital output can be shared with other - QM instances. + shareable (bool, deprecated): If True, the digital output can be shared + with other QM instances. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusDigitalOutputPort(shareable=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. inverted (bool, deprecated): If True, the digital output is inverted. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusDigitalOutputPort(inverted=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. .""" opx_output: Union[Tuple[str, int], Tuple[str, int, int], DigitalOutputPort] - delay: int = None - buffer: int = None + delay: Optional[int] = None + buffer: Optional[int] = None - shareable: bool = None - inverted: bool = None + shareable: Optional[bool] = None + inverted: Optional[bool] = None def generate_element_config(self) -> Dict[str, int]: """Generates the config entry for a digital channel in the QUA config. @@ -180,7 +182,7 @@ def generate_element_config(self) -> Dict[str, int]: if isinstance(self.opx_output, DigitalOutputPort): opx_output = self.opx_output.port_tuple else: - opx_output = tuple(self.opx_output) + opx_output = tuple(self.opx_output) # type: ignore[assignment] digital_cfg: Dict[str, Any] = {"port": opx_output} if self.delay is not None: @@ -195,7 +197,7 @@ def apply_to_config(self, config: dict) -> None: config.controllers..digital_outputs. will be updated with the shareable and inverted settings of this channel if specified. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ if isinstance(self.opx_output, DigitalOutputPort): @@ -238,12 +240,13 @@ def apply_to_config(self, config: dict) -> None: shareable = self.shareable if self.shareable is not None else False inverted = self.inverted if self.inverted is not None else False + digital_output_port: DigitalOutputPort if len(self.opx_output) == 2: - digital_output_port = OPXPlusDigitalOutputPort( + digital_output_port = OPXPlusDigitalOutputPort( # type: ignore[misc] *self.opx_output, shareable=shareable, inverted=inverted ) else: - digital_output_port = FEMDigitalOutputPort( + digital_output_port = FEMDigitalOutputPort( # type: ignore[misc] *self.opx_output, shareable=shareable, inverted=inverted ) digital_output_port.apply_to_config(config) @@ -274,7 +277,7 @@ def channel(self) -> Optional["Channel"]: if isinstance(self.parent, Channel): return self.parent else: - return + return None @property def config_settings(self): @@ -309,7 +312,7 @@ class TimeTaggingAddon(QuamComponent): derivative_polarity (Literal["above", "below"]): The polarity of the derivative threshold. Default is "below". - For details see [Time Tagging](https://docs.quantum-machines.co/latest/docs/Guides/features/#time-tagging) + For details see [Time Tagging](https://docs.quantum-machines.co/latest/docs/Guides/features/#time-tagging) # noqa: E501 """ signal_threshold: float = 800 / 4096 @@ -324,7 +327,7 @@ def channel(self) -> Optional["Channel"]: if isinstance(self.parent, Channel): return self.parent else: - return + return None @property def config_settings(self): @@ -377,7 +380,7 @@ class Channel(QuamComponent, ABC): operations: Dict[str, Pulse] = field(default_factory=dict) - id: Union[str, int] = None + id: Optional[Union[str, int]] = None _default_label: ClassVar[str] = "ch" # Used to determine name from id digital_outputs: Dict[str, DigitalOutputChannel] = field(default_factory=dict) @@ -392,7 +395,7 @@ def name(self) -> str: cls_name = self.__class__.__name__ if self.id is not None: - if str_ref.is_reference(self.id): + if isinstance(self.id, str) and str_ref.is_reference(self.id): raise AttributeError( f"{cls_name}.name cannot be determined. " f"Please either set {cls_name}.id to a string or integer, " @@ -410,7 +413,7 @@ def name(self) -> str: "a name." ) if isinstance(self.parent, QuamDict): - return self.inferred_id + return str(self.inferred_id) if not hasattr(self.parent, "name"): raise AttributeError( f"{cls_name}.name cannot be determined. " @@ -437,11 +440,11 @@ def play( self, pulse_name: str, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - duration: ScalarInt = None, - condition: ScalarBool = None, - chirp: ChirpType = None, - truncate: ScalarInt = None, - timestamp_stream: StreamType = None, + duration: Optional[ScalarInt] = None, + condition: Optional[ScalarBool] = None, + chirp: Optional[ChirpType] = None, + truncate: Optional[ScalarInt] = None, + timestamp_stream: Optional[StreamType] = None, continue_chirp: bool = False, target: str = "", validate: bool = True, @@ -548,7 +551,8 @@ def ramp_to_zero(self, duration: Optional[int] = None): qua.ramp_to_zero(self.name, duration=duration) def wait(self, duration: ScalarInt, *other_elements: Union[str, "Channel"]): - """Wait for the given duration on all provided elements without outputting anything. + """Wait for the given duration on all provided elements without + outputting anything. Duration is in units of the clock cycle (4ns) @@ -559,17 +563,19 @@ def wait(self, duration: ScalarInt, *other_elements: Union[str, "Channel"]): in addition to this channel Warning: - In case the value of this is outside the range above, unexpected results may occur. + In case the value of this is outside the range above, unexpected + results may occur. Note: The current channel element is always included in the wait operation. Note: - The purpose of the `wait` operation is to add latency. In most cases, the - latency added will be exactly the same as that specified by the QUA variable or - the literal used. However, in some cases an additional computational latency may - be added. If the actual wait time has significance, such as in characterization - experiments, the actual wait time should always be verified with a simulator. + The purpose of the `wait` operation is to add latency. In most + cases, the latency added will be exactly the same as that specified + by the QUA variable or the literal used. However, in some cases an + additional computational latency may be added. If the actual wait + time has significance, such as in characterization experiments, the + actual wait time should always be verified with a simulator. """ other_elements_str = [ element if isinstance(element, str) else str(element) @@ -741,6 +747,7 @@ def apply_to_config(self, config: Dict[str, dict]) -> None: ) qua_below_1_2_2 = True + core: Optional[str] if self.core is not None and self.thread is not None: warnings.warn( "The 'thread' and 'core' arguments are mutually exclusive. " @@ -781,30 +788,33 @@ class SingleChannel(Channel): filter_fir_taps (List[float], deprecated): FIR filter taps for the output port. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogOutputPort(feedforward_filter=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. filter_iir_taps (List[float], deprecated): IIR filter taps for the output port. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogOutputPort(feedback_filter=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. opx_output_offset (float, deprecated): DC offset for the output port. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogOutputPort(offset=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. intermediate_frequency (float): Intermediate frequency of OPX output, default is None. """ opx_output: LF_output_port_types - filter_fir_taps: List[float] = None - filter_iir_taps: List[float] = None + filter_fir_taps: Optional[List[float]] = None + filter_iir_taps: Optional[List[float]] = None - opx_output_offset: float = None + opx_output_offset: Optional[float] = None @property def sampling_rate(self) -> float: """Sampling rate from the output port, defaulting to 1 GHz.""" if hasattr(self.opx_output, "sampling_rate"): - return self.opx_output.sampling_rate + return self.opx_output.sampling_rate # type: ignore[union-attr] return 1e9 def set_dc_offset(self, offset: ScalarFloat): @@ -821,7 +831,7 @@ def set_dc_offset(self, offset: ScalarFloat): def apply_to_config(self, config: dict): """Adds this SingleChannel to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ # Add pulses & waveforms @@ -871,7 +881,7 @@ def apply_to_config(self, config: dict): if isinstance(self.opx_output, LFAnalogOutputPort): opx_port = self.opx_output elif len(self.opx_output) == 2: - opx_port = OPXPlusAnalogOutputPort( + opx_port = OPXPlusAnalogOutputPort( # type: ignore[misc] *self.opx_output, offset=self.opx_output_offset, feedforward_filter=filter_fir_taps, @@ -879,7 +889,7 @@ def apply_to_config(self, config: dict): ) opx_port.apply_to_config(config) else: - opx_port = LFFEMAnalogOutputPort( + opx_port = LFFEMAnalogOutputPort( # type: ignore[misc] *self.opx_output, offset=self.opx_output_offset, feedforward_filter=filter_fir_taps, @@ -905,7 +915,8 @@ class InSingleChannel(Channel): opx_input_offset (float, deprecated): DC offset for the input port. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogInputPort(offset=...)` instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. intermediate_frequency (float): Intermediate frequency of OPX input, default is None. time_of_flight (int): Round-trip signal duration in nanoseconds. @@ -914,7 +925,7 @@ class InSingleChannel(Channel): """ opx_input: LF_input_port_types - opx_input_offset: float = None + opx_input_offset: Optional[float] = None time_of_flight: int = 140 smearing: int = 0 @@ -924,7 +935,7 @@ class InSingleChannel(Channel): def apply_to_config(self, config: dict): """Adds this InSingleChannel to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ # Add output to config @@ -948,12 +959,12 @@ def apply_to_config(self, config: dict): if isinstance(self.opx_input, LFAnalogInputPort): opx_port = self.opx_input elif len(self.opx_input) == 2: - opx_port = OPXPlusAnalogInputPort( + opx_port = OPXPlusAnalogInputPort( # type: ignore[misc] *self.opx_input, offset=self.opx_input_offset ) opx_port.apply_to_config(config) else: - opx_port = LFFEMAnalogInputPort( + opx_port = LFFEMAnalogInputPort( # type: ignore[misc] *self.opx_input, offset=self.opx_input_offset ) opx_port.apply_to_config(config) @@ -964,7 +975,7 @@ def measure( self, pulse_name: str, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - qua_vars: Tuple[QuaVariableFloat, ...] = None, + qua_vars: Optional[Tuple[QuaVariableFloat, ...]] = None, stream=None, ) -> Tuple[QuaVariableFloat, QuaVariableFloat]: """Perform a full demodulation measurement on this channel. @@ -991,7 +1002,11 @@ def measure( variables. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if qua_vars is not None: if not isinstance(qua_vars, Sequence) or len(qua_vars) != 2: @@ -1000,7 +1015,9 @@ def measure( f"which is not a tuple of two QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed) for _ in range(2)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed) for _ in range(2) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1010,19 +1027,23 @@ def measure( qua.measure( pulse_name_with_amp_scale, self.name, - qua.demod.full(integration_weight_labels[0], qua_vars[0], "out1"), - qua.demod.full(integration_weight_labels[1], qua_vars[1], "out1"), + qua.demod.full( + integration_weight_labels[0], qua_vars[0], "out1" # type: ignore[index] + ), + qua.demod.full( + integration_weight_labels[1], qua_vars[1], "out1" # type: ignore[index] + ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] def measure_accumulated( self, pulse_name: str, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - num_segments: int = None, - segment_length: int = None, - qua_vars: Tuple[QuaVariableFloat, ...] = None, + num_segments: Optional[int] = None, + segment_length: Optional[int] = None, + qua_vars: Optional[Tuple[QuaVariableFloat, ...]] = None, stream=None, ) -> Tuple[QuaVariableFloat, QuaVariableFloat]: """Perform an accumulated demodulation measurement on this channel. @@ -1054,11 +1075,16 @@ def measure_accumulated( ValueError: If `qua_vars` is provided and is not a tuple of two QUA variables. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if num_segments is None and segment_length is None: raise ValueError( - "InOutSingleChannel.measure_accumulated requires either 'segment_length' " + "InOutSingleChannel.measure_accumulated requires either " + "'segment_length' " "or 'num_segments' to be provided." ) elif num_segments is not None and segment_length is not None: @@ -1066,9 +1092,11 @@ def measure_accumulated( "InOutSingleChannel.measure_accumulated received both 'segment_length' " "and 'num_segments'. Please provide only one." ) - elif num_segments is None: - num_segments = int(pulse.length / (4 * segment_length)) # Number of slices - elif segment_length is None: + elif segment_length is not None: + num_segments = int(pulse.length / (4 * segment_length)) + else: + if num_segments is None: + raise ValueError("Expected num_segments to be set") segment_length = int(pulse.length / (4 * num_segments)) if qua_vars is not None: @@ -1078,7 +1106,9 @@ def measure_accumulated( f"which is not a tuple of two QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed, size=num_segments) for _ in range(2)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed, size=num_segments) for _ in range(2) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1089,22 +1119,22 @@ def measure_accumulated( pulse_name_with_amp_scale, self.name, qua.demod.accumulated( - integration_weight_labels[0], qua_vars[0], segment_length, "out1" + integration_weight_labels[0], qua_vars[0], segment_length, "out1" # type: ignore[index,arg-type] # noqa: E501 ), qua.demod.accumulated( - integration_weight_labels[1], qua_vars[1], segment_length, "out1" + integration_weight_labels[1], qua_vars[1], segment_length, "out1" # type: ignore[index,arg-type] # noqa: E501 ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] def measure_sliced( self, pulse_name: str, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - num_segments: int = None, - segment_length: int = None, - qua_vars: Tuple[QuaVariableFloat, ...] = None, + num_segments: Optional[int] = None, + segment_length: Optional[int] = None, + qua_vars: Optional[Tuple[QuaVariableFloat, ...]] = None, stream=None, ) -> Tuple[QuaVariableFloat, QuaVariableFloat]: """Perform an accumulated demodulation measurement on this channel. @@ -1136,7 +1166,11 @@ def measure_sliced( ValueError: If `qua_vars` is provided and is not a tuple of two QUA variables. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if num_segments is None and segment_length is None: raise ValueError( @@ -1148,9 +1182,11 @@ def measure_sliced( "InOutSingleChannel.measure_sliced received both 'segment_length' " "and 'num_segments'. Please provide only one." ) - elif num_segments is None: - num_segments = int(pulse.length / (4 * segment_length)) # Number of slices - elif segment_length is None: + elif segment_length is not None: + num_segments = int(pulse.length / (4 * segment_length)) + else: + if num_segments is None: + raise ValueError("Expected num_segments to be set") segment_length = int(pulse.length / (4 * num_segments)) if qua_vars is not None: @@ -1160,7 +1196,9 @@ def measure_sliced( f"which is not a tuple of two QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed, size=num_segments) for _ in range(2)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed, size=num_segments) for _ in range(2) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1171,14 +1209,14 @@ def measure_sliced( pulse_name_with_amp_scale, self.name, qua.demod.sliced( - integration_weight_labels[0], qua_vars[0], segment_length, "out1" + integration_weight_labels[0], qua_vars[0], segment_length, "out1" # type: ignore[index,arg-type] # noqa: E501 ), qua.demod.sliced( - integration_weight_labels[1], qua_vars[1], segment_length, "out1" + integration_weight_labels[1], qua_vars[1], segment_length, "out1" # type: ignore[index,arg-type] # noqa: E501 ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] def measure_time_tagging( self, @@ -1191,7 +1229,7 @@ def measure_time_tagging( ) -> Tuple[QuaVariableInt, QuaScalarInt]: """Perform a time tagging measurement on this channel. - For details see https://docs.quantum-machines.co/latest/docs/Guides/features/#time-tagging + For details see https://docs.quantum-machines.co/latest/docs/Guides/features/#time-tagging # noqa: E501 Args: pulse_name (str): The name of the pulse to play. Should be registered in @@ -1219,9 +1257,9 @@ def measure_time_tagging( if mode == "analog": time_tagging_func = qua.time_tagging.analog elif mode == "high_res": - time_tagging_func = qua.time_tagging.high_res + time_tagging_func = qua.time_tagging.high_res # type: ignore[assignment] elif mode == "digital": - time_tagging_func = qua.time_tagging.digital + time_tagging_func = qua.time_tagging.digital # type: ignore[assignment] else: raise ValueError(f"Invalid time tagging mode: {mode}") @@ -1229,7 +1267,7 @@ def measure_time_tagging( times = qua.declare(int, size=size) counts = qua.declare(int) else: - times, counts = qua_vars + times, counts = qua_vars # type: ignore[assignment] qua.measure( pulse_name, @@ -1237,7 +1275,7 @@ def measure_time_tagging( time_tagging_func(target=times, max_time=max_time, targetLen=counts), adc_stream=stream, ) - return times, counts + return times, counts # type: ignore[return-value] def _raise_inferred_freq_error( @@ -1290,7 +1328,7 @@ def inferred_RF_frequency(self) -> float: "intermediate_frequency", self.intermediate_frequency, ) - return self.LO_frequency + self.intermediate_frequency + return self.LO_frequency + self.intermediate_frequency # type: ignore[operator] @property def inferred_intermediate_frequency(self) -> float: @@ -1334,7 +1372,7 @@ def inferred_LO_frequency(self) -> float: "intermediate_frequency", self.intermediate_frequency, ) - return self.RF_frequency - self.intermediate_frequency + return self.RF_frequency - self.intermediate_frequency # type: ignore[operator] @quam_dataclass @@ -1354,11 +1392,13 @@ class IQChannel(_OutComplexChannel): opx_output_offset_I (float, deprecated): The offset of the I channel. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogOutputPort(offset=...)` on the I port instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. opx_output_offset_Q (float, deprecated): The offset of the Q channel. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogOutputPort(offset=...)` on the Q port instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. intermediate_frequency (float): Intermediate frequency of the mixer. Default is 0.0 LO_frequency (float): Local oscillator frequency. Default is the LO frequency @@ -1372,13 +1412,13 @@ class IQChannel(_OutComplexChannel): opx_output_I: LF_output_port_types opx_output_Q: LF_output_port_types - opx_output_offset_I: float = None - opx_output_offset_Q: float = None + opx_output_offset_I: Optional[float] = None + opx_output_offset_Q: Optional[float] = None frequency_converter_up: BaseFrequencyConverter - LO_frequency: float = "#./frequency_converter_up/LO_frequency" - RF_frequency: float = "#./inferred_RF_frequency" + LO_frequency: float = "#./frequency_converter_up/LO_frequency" # type: ignore[assignment] + RF_frequency: float = "#./inferred_RF_frequency" # type: ignore[assignment] _default_label: ClassVar[str] = "IQ" @@ -1386,7 +1426,7 @@ class IQChannel(_OutComplexChannel): def sampling_rate(self) -> float: """Sampling rate from the I output port, defaulting to 1 GHz.""" if hasattr(self.opx_output_I, "sampling_rate"): - return self.opx_output_I.sampling_rate + return self.opx_output_I.sampling_rate # type: ignore[union-attr] return 1e9 @property @@ -1428,7 +1468,7 @@ def set_dc_offset(self, offset: ScalarFloat, element_input: Literal["I", "Q"]): def apply_to_config(self, config: dict): """Adds this IQChannel to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ # Add pulses & waveforms @@ -1476,7 +1516,9 @@ def apply_to_config(self, config: dict): element_config["RF_inputs"] = { "port": (octave.name, self.frequency_converter_up.id) } - elif str_ref.is_reference(self.frequency_converter_up): + elif isinstance(self.frequency_converter_up, str) and str_ref.is_reference( + self.frequency_converter_up + ): raise ValueError( f"Error generating config: channel {self.name} could not determine " f'"frequency_converter_up", it seems to point to a non-existent ' @@ -1497,10 +1539,10 @@ def apply_to_config(self, config: dict): if isinstance(opx_output, LFAnalogOutputPort): opx_port = opx_output elif len(opx_output) == 2: - opx_port = OPXPlusAnalogOutputPort(*opx_output, offset=offset) + opx_port = OPXPlusAnalogOutputPort(*opx_output, offset=offset) # type: ignore[misc] opx_port.apply_to_config(config) else: - opx_port = LFFEMAnalogOutputPort(*opx_output, offset=offset) + opx_port = LFFEMAnalogOutputPort(*opx_output, offset=offset) # type: ignore[misc] opx_port.apply_to_config(config) if "mixInputs" in element_config: @@ -1524,7 +1566,7 @@ def measure( self, pulse_name: str, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - qua_vars: Tuple[QuaVariableFloat, QuaVariableFloat] = None, + qua_vars: Optional[Tuple[QuaVariableFloat, QuaVariableFloat]] = None, stream=None, ) -> Tuple[QuaVariableFloat, QuaVariableFloat]: """Perform a full dual demodulation measurement on this channel. @@ -1546,7 +1588,11 @@ def measure( If provided as input, the same variables will be returned. If not provided, new variables will be declared and returned. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if qua_vars is not None: if not isinstance(qua_vars, Sequence) or len(qua_vars) != 2: @@ -1555,7 +1601,9 @@ def measure( f"tuple of two QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed) for _ in range(2)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed) for _ in range(2) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1570,18 +1618,18 @@ def measure( element_output1="out1", iw2=integration_weight_labels[1], element_output2="out2", - target=qua_vars[0], + target=qua_vars[0], # type: ignore[index] ), qua.dual_demod.full( iw1=integration_weight_labels[2], element_output1="out1", iw2=integration_weight_labels[0], element_output2="out2", - target=qua_vars[1], + target=qua_vars[1], # type: ignore[index] ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] def measure_accumulated( self, @@ -1619,11 +1667,16 @@ def measure_accumulated( If provided as input, the same variables will be returned. If not provided, new variables will be declared and returned. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if num_segments is None and segment_length is None: raise ValueError( - "InOutSingleChannel.measure_accumulated requires either 'segment_length' " + "InOutSingleChannel.measure_accumulated requires either " + "'segment_length' " "or 'num_segments' to be provided." ) elif num_segments is not None and segment_length is not None: @@ -1631,9 +1684,11 @@ def measure_accumulated( "InOutSingleChannel.measure_accumulated received both 'segment_length' " "and 'num_segments'. Please provide only one." ) - elif num_segments is None: - num_segments = int(pulse.length / (4 * segment_length)) # Number of slices - elif segment_length is None: + elif segment_length is not None: + num_segments = int(pulse.length / (4 * segment_length)) + else: + if num_segments is None: + raise ValueError("Expected num_segments to be set") segment_length = int(pulse.length / (4 * num_segments)) if qua_vars is not None: @@ -1643,7 +1698,9 @@ def measure_accumulated( f"which is not a tuple of four QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed, size=num_segments) for _ in range(4)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed, size=num_segments) for _ in range(4) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1654,20 +1711,20 @@ def measure_accumulated( pulse_name_with_amp_scale, self.name, qua.demod.accumulated( - integration_weight_labels[0], qua_vars[0], segment_length, "out1" + integration_weight_labels[0], qua_vars[0], segment_length, "out1" # type: ignore[index,arg-type] ), qua.demod.accumulated( - integration_weight_labels[1], qua_vars[1], segment_length, "out2" + integration_weight_labels[1], qua_vars[1], segment_length, "out2" # type: ignore[index,arg-type] ), qua.demod.accumulated( - integration_weight_labels[2], qua_vars[2], segment_length, "out1" + integration_weight_labels[2], qua_vars[2], segment_length, "out1" # type: ignore[index,arg-type] ), qua.demod.accumulated( - integration_weight_labels[0], qua_vars[3], segment_length, "out2" + integration_weight_labels[0], qua_vars[3], segment_length, "out2" # type: ignore[index,arg-type] ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] def measure_sliced( self, @@ -1705,7 +1762,11 @@ def measure_sliced( If provided as input, the same variables will be returned. If not provided, new variables will be declared and returned. """ - pulse: BaseReadoutPulse = self.operations[pulse_name] + pulse = self.operations[pulse_name] + if not isinstance(pulse, BaseReadoutPulse): + raise ValueError( + f"Pulse '{pulse_name}' is not a BaseReadoutPulse, got {type(pulse)}" + ) if num_segments is None and segment_length is None: raise ValueError( @@ -1717,9 +1778,11 @@ def measure_sliced( "InOutSingleChannel.measure_sliced received both 'segment_length' " "and 'num_segments'. Please provide only one." ) - elif num_segments is None: - num_segments = int(pulse.length / (4 * segment_length)) # Number of slices - elif segment_length is None: + elif segment_length is not None: + num_segments = int(pulse.length / (4 * segment_length)) + else: + if num_segments is None: + raise ValueError("Expected num_segments to be set") segment_length = int(pulse.length / (4 * num_segments)) if qua_vars is not None: @@ -1729,7 +1792,9 @@ def measure_sliced( f"which is not a tuple of four QUA variables. Received {qua_vars=}" ) else: - qua_vars = [qua.declare(qua.fixed, size=num_segments) for _ in range(4)] + qua_vars = [ # type: ignore[assignment] + qua.declare(qua.fixed, size=num_segments) for _ in range(4) + ] pulse_name_with_amp_scale = add_amplitude_scale_to_pulse_name( pulse_name, amplitude_scale @@ -1740,20 +1805,20 @@ def measure_sliced( pulse_name_with_amp_scale, self.name, qua.demod.sliced( - integration_weight_labels[0], qua_vars[0], segment_length, "out1" + integration_weight_labels[0], qua_vars[0], segment_length, "out1" # type: ignore[index,arg-type] ), qua.demod.sliced( - integration_weight_labels[1], qua_vars[1], segment_length, "out2" + integration_weight_labels[1], qua_vars[1], segment_length, "out2" # type: ignore[index,arg-type] ), qua.demod.sliced( - integration_weight_labels[2], qua_vars[2], segment_length, "out1" + integration_weight_labels[2], qua_vars[2], segment_length, "out1" # type: ignore[index,arg-type] ), qua.demod.sliced( - integration_weight_labels[0], qua_vars[3], segment_length, "out2" + integration_weight_labels[0], qua_vars[3], segment_length, "out2" # type: ignore[index,arg-type] ), adc_stream=stream, ) - return tuple(qua_vars) + return tuple(qua_vars) # type: ignore[return-value,arg-type] @quam_dataclass @@ -1774,11 +1839,13 @@ class InIQChannel(_InComplexChannel): opx_input_offset_I (float, deprecated): The offset of the I channel. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogInputPort(offset=...)` on the I port instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. opx_input_offset_Q (float, deprecated): The offset of the Q channel. **Deprecated**: This property has been moved to Port objects. Use `OPXPlusAnalogInputPort(offset=...)` on the Q port instead. - Will be removed in v0.6.0. See [Port documentation](channel-ports.md) for details. + Will be removed in v0.6.0. + See [Port documentation](channel-ports.md) for details. frequency_converter_down (Optional[FrequencyConverter]): Frequency converter QUAM component for the IQ input port. Only needed for the old Octave. time_of_flight (int): Round-trip signal duration in nanoseconds. @@ -1793,19 +1860,19 @@ class InIQChannel(_InComplexChannel): time_of_flight: int = 140 smearing: int = 0 - opx_input_offset_I: float = None - opx_input_offset_Q: float = None + opx_input_offset_I: Optional[float] = None + opx_input_offset_Q: Optional[float] = None input_gain: Optional[int] = None - frequency_converter_down: BaseFrequencyConverter = None + frequency_converter_down: Optional[BaseFrequencyConverter] = None _default_label: ClassVar[str] = "IQ" def apply_to_config(self, config: dict): """Adds this InOutIQChannel to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ super().apply_to_config(config) @@ -1828,7 +1895,9 @@ def apply_to_config(self, config: dict): element_config["RF_outputs"] = { "port": (octave.name, self.frequency_converter_down.id) } - elif str_ref.is_reference(self.frequency_converter_down): + elif isinstance(self.frequency_converter_down, str) and str_ref.is_reference( + self.frequency_converter_down + ): raise ValueError( f"Error generating config: channel {self.name} could not determine " f'"frequency_converter_down", it seems to point to a non-existent ' @@ -1864,12 +1933,12 @@ def apply_to_config(self, config: dict): if isinstance(opx_input, LFAnalogInputPort): opx_port = opx_input elif len(opx_input) == 2: - opx_port = OPXPlusAnalogInputPort( + opx_port = OPXPlusAnalogInputPort( # type: ignore[misc] *opx_input, offset=offset, gain_db=input_gain ) opx_port.apply_to_config(config) else: - opx_port = LFFEMAnalogInputPort( + opx_port = LFFEMAnalogInputPort( # type: ignore[misc] *opx_input, offset=offset, gain_db=input_gain ) opx_port.apply_to_config(config) @@ -2039,8 +2108,8 @@ class MWChannel(_OutComplexChannel): opx_output: MWFEMAnalogOutputPort upconverter: int = 1 - LO_frequency: float = "#./upconverter_frequency" - RF_frequency: float = "#./inferred_RF_frequency" + LO_frequency: float = "#./upconverter_frequency" # type: ignore[assignment] + RF_frequency: float = "#./inferred_RF_frequency" # type: ignore[assignment] @property def sampling_rate(self) -> float: diff --git a/quam/components/hardware.py b/quam/components/hardware.py index 6b557e79..4bbee6ed 100644 --- a/quam/components/hardware.py +++ b/quam/components/hardware.py @@ -1,5 +1,5 @@ import numpy as np -from typing import List +from typing import List, Optional, Union from quam.core import QuamComponent, quam_dataclass from quam.utils import string_reference as str_ref @@ -22,8 +22,8 @@ class LocalOscillator(QuamComponent): Not used for the QUA configuration """ - frequency: float = None - power: float = None + frequency: Optional[float] = None + power: Optional[float] = None def configure(self): ... @@ -48,8 +48,8 @@ class Mixer(QuamComponent): correction_phase (float, optional): The phase imbalance of the mixer in radians. """ - local_oscillator_frequency: float = "#../local_oscillator/frequency" - intermediate_frequency: float = "#../../intermediate_frequency" + local_oscillator_frequency: Union[float, str] = "#../local_oscillator/frequency" + intermediate_frequency: Union[float, str] = "#../../intermediate_frequency" correction_gain: float = 0 correction_phase: float = 0 @@ -131,9 +131,9 @@ class FrequencyConverter(BaseFrequencyConverter): gain (float): The gain of the frequency converter. """ - local_oscillator: LocalOscillator = None - mixer: Mixer = None - gain: float = None + local_oscillator: Optional[LocalOscillator] = None + mixer: Optional[Mixer] = None + gain: Optional[float] = None @property def LO_frequency(self): diff --git a/quam/components/macro/__init__.py b/quam/components/macro/__init__.py index a3617026..8954645a 100644 --- a/quam/components/macro/__init__.py +++ b/quam/components/macro/__init__.py @@ -1,5 +1,6 @@ from .qubit_macros import * from .qubit_pair_macros import * +from . import qubit_macros, qubit_pair_macros __all__ = [ *qubit_macros.__all__, diff --git a/quam/components/macro/qubit_macros.py b/quam/components/macro/qubit_macros.py index 6e1b32b8..d50b5aa8 100644 --- a/quam/components/macro/qubit_macros.py +++ b/quam/components/macro/qubit_macros.py @@ -1,5 +1,5 @@ from abc import ABC -from typing import Optional, Union, List +from typing import Union from quam.core.macro import QuamMacro from quam.components.pulses import Pulse from quam.core import quam_dataclass diff --git a/quam/components/octave.py b/quam/components/octave.py index 782bb0bb..57494ecf 100644 --- a/quam/components/octave.py +++ b/quam/components/octave.py @@ -55,7 +55,7 @@ class Octave(QuamComponent): """ name: str - calibration_db_path: str = None + calibration_db_path: Optional[str] = None RF_outputs: Dict[int, "OctaveUpConverter"] = field(default_factory=dict) RF_inputs: Dict[int, "OctaveDownConverter"] = field(default_factory=dict) @@ -147,7 +147,7 @@ class OctaveFrequencyConverter(BaseFrequencyConverter, ABC): """ id: int - channel: Channel = None + channel: Optional[Channel] = None @property def octave(self) -> Optional[Octave]: @@ -159,7 +159,7 @@ def octave(self) -> Optional[Octave]: return parent_parent @property - def config_settings(self) -> Dict[str, Any]: + def config_settings(self) -> Dict[str, Any]: # type: ignore[override] """Specifies that the converter will be added to the config after the Octave.""" return {"after": [self.octave]} @@ -222,7 +222,7 @@ class OctaveUpConverter(OctaveFrequencyConverter): entering the mixer. Off by default. """ - LO_frequency: float = None + LO_frequency: Optional[float] = None LO_source: Literal["internal", "external"] = "internal" gain: float = 0 output_mode: Literal[ @@ -257,6 +257,7 @@ def apply_to_config(self, config: Dict) -> None: ) super().apply_to_config(config) + assert self.octave is not None if self.id in config["octaves"][self.octave.name]["RF_outputs"]: raise KeyError( @@ -265,13 +266,14 @@ def apply_to_config(self, config: Dict) -> None: f'already has an entry for OctaveDownConverter with id "{self.id}"' ) - output_config = config["octaves"][self.octave.name]["RF_outputs"][self.id] = { + output_config: Dict[str, Any] = { "LO_frequency": self.LO_frequency, "LO_source": self.LO_source, "gain": self.gain, "output_mode": self.output_mode, "input_attenuators": self.input_attenuators, } + config["octaves"][self.octave.name]["RF_outputs"][self.id] = output_config if isinstance(self.channel, SingleChannel): if isinstance(self.channel.opx_output, LFAnalogOutputPort): output_config["I_connection"] = self.channel.opx_output.port_tuple @@ -320,7 +322,7 @@ class OctaveDownConverter(OctaveFrequencyConverter): are connected to the opposite OPX inputs. """ - LO_frequency: float = None + LO_frequency: Optional[float] = None LO_source: Literal["internal", "external"] = "internal" IF_mode_I: Literal["direct", "envelope", "mixer", "off"] = "direct" IF_mode_Q: Literal["direct", "envelope", "mixer", "off"] = "direct" @@ -357,6 +359,7 @@ def apply_to_config(self, config: Dict) -> None: return super().apply_to_config(config) + assert self.octave is not None # Add IF_outputs (physical wiring) whenever channel is connected. # This is required for qm.calibrate_element() to find calibration connections, @@ -420,16 +423,16 @@ class OctaveOld(QuamComponent): port: int qmm_host: str qmm_port: int - connection_headers: Dict[str, str] = None - connectivity: str = None + connection_headers: Optional[Dict[str, str]] = None + connectivity: Optional[str] = None - calibration_db: str = None + calibration_db: Optional[str] = None - octave_config: ClassVar[QmOctaveConfig] = None + octave_config: ClassVar[Optional[QmOctaveConfig]] = None _qms: ClassVar[Dict[str, QuantumMachinesManager]] = {} - _qm: ClassVar[QuantumMachine] = None - octave: ClassVar[QmOctave] = None - _channel_to_qe: ClassVar[dict] = None + _qm: ClassVar[Optional[QuantumMachine]] = None + octave: ClassVar[Optional[QmOctave]] = None + _channel_to_qe: ClassVar[Optional[dict]] = None def _initialize_config(self): calibration_db = self.calibration_db @@ -452,7 +455,10 @@ def _initialize_qm(self) -> QuantumMachine: octave=self.octave_config, connection_headers=self.connection_headers, ) - qm = qmm.open_qm(self.get_root().generate_config()) + root = self.get_root() + assert root is not None + qm = qmm.open_qm(root.generate_config()) + assert isinstance(qm, QuantumMachine) return qm def get_portmap(self): @@ -508,6 +514,8 @@ def configure(self): self.configure_octave_settings() def calibrate(self, channel: str, lo_freq: int, if_freq: int, gain: float): + assert self._channel_to_qe is not None + assert self.octave is not None channel_qe = self._channel_to_qe[self.name, channel] self.octave.set_lo_frequency(channel_qe, lo_freq) self.octave.set_rf_output_gain(channel_qe, gain) @@ -516,10 +524,14 @@ def calibrate(self, channel: str, lo_freq: int, if_freq: int, gain: float): self.octave.calibrate_element(channel_qe, [(lo_freq, if_freq)]) def set_frequency(self, channel: str, frequency: float): + assert self._channel_to_qe is not None + assert self.octave is not None channel_qe = self._channel_to_qe[self.name, channel] self.octave.set_lo_frequency(channel_qe, frequency) def set_gain(self, channel: str, gain: float): + assert self._channel_to_qe is not None + assert self.octave is not None channel_qe = self._channel_to_qe[self.name, channel] self.octave.set_rf_output_gain(channel_qe, gain) diff --git a/quam/components/ports/__init__.py b/quam/components/ports/__init__.py index 1aa4d70c..143b0dd4 100644 --- a/quam/components/ports/__init__.py +++ b/quam/components/ports/__init__.py @@ -4,6 +4,14 @@ from .digital_inputs import * from .digital_outputs import * from .ports_containers import * +from . import ( + analog_outputs, + analog_inputs, + base_ports, + digital_inputs, + digital_outputs, + ports_containers, +) __all__ = [ *analog_outputs.__all__, diff --git a/quam/components/ports/analog_outputs.py b/quam/components/ports/analog_outputs.py index a9d112d0..ea79b63a 100644 --- a/quam/components/ports/analog_outputs.py +++ b/quam/components/ports/analog_outputs.py @@ -116,7 +116,7 @@ def __post_init__(self) -> None: ) def get_port_properties(self) -> Dict[str, Any]: - port_cfg = { + port_cfg: Dict[str, Any] = { "band": self.band, "delay": self.delay, "shareable": self.shareable, diff --git a/quam/components/ports/base_ports.py b/quam/components/ports/base_ports.py index 399bbdc5..2bd8d1f9 100644 --- a/quam/components/ports/base_ports.py +++ b/quam/components/ports/base_ports.py @@ -23,7 +23,9 @@ def get_port_properties(self) -> Dict[str, Any]: @property @abstractmethod - def port_tuple(self) -> Union[Tuple[str, int], Tuple[str, int, int]]: + def port_tuple( + self, + ) -> Union[Tuple[Union[str, int], int], Tuple[Union[str, int], int, int]]: pass def _update_port_config(self, port_config, port_properties): diff --git a/quam/components/ports/ports_containers.py b/quam/components/ports/ports_containers.py index f168c9b0..2641770a 100644 --- a/quam/components/ports/ports_containers.py +++ b/quam/components/ports/ports_containers.py @@ -1,6 +1,5 @@ -from typing import ClassVar, Dict, Optional, TypeVar, Union +from typing import ClassVar, Dict, Optional, Union from dataclasses import field -from quam.components.ports.base_ports import FEMPort from quam.core import quam_dataclass, QuamComponent from quam.core.quam_classes import QuamBase from .analog_outputs import ( @@ -79,13 +78,13 @@ def _get_port( ports = controllers[controller_id] if port_type == "analog_output": - ports[port_id] = OPXPlusAnalogOutputPort(controller_id, port_id, **kwargs) + ports[port_id] = OPXPlusAnalogOutputPort(controller_id, port_id, **kwargs) # type: ignore[misc] elif port_type == "analog_input": - ports[port_id] = OPXPlusAnalogInputPort(controller_id, port_id, **kwargs) + ports[port_id] = OPXPlusAnalogInputPort(controller_id, port_id, **kwargs) # type: ignore[misc] elif port_type == "digital_output": - ports[port_id] = OPXPlusDigitalOutputPort(controller_id, port_id, **kwargs) + ports[port_id] = OPXPlusDigitalOutputPort(controller_id, port_id, **kwargs) # type: ignore[misc] elif port_type == "digital_input": - ports[port_id] = OPXPlusDigitalInputPort(controller_id, port_id, **kwargs) + ports[port_id] = OPXPlusDigitalInputPort(controller_id, port_id, **kwargs) # type: ignore[misc] return ports[port_id] @@ -103,12 +102,15 @@ def reference_to_port( try: elems = port_reference.split("/") - port_type, controller_id, port_id = elems[-3:] + port_type_str, controller_id_str, port_id_str = elems[-3:] - port_type = port_type[:-1] - if controller_id.isdigit(): - controller_id = int(controller_id) - port_id = int(port_id) + port_type = port_type_str[:-1] + controller_id: Union[str, int] = ( + int(controller_id_str) + if controller_id_str.isdigit() + else controller_id_str + ) + port_id: int = int(port_id_str) except Exception as e: raise ValueError( f"Unable to parse port reference for OPX+: {port_reference}" @@ -215,11 +217,11 @@ def _get_port( ports = fems[fem_id] if port_type == "analog_output": - ports[port_id] = LFFEMAnalogOutputPort( + ports[port_id] = LFFEMAnalogOutputPort( # type: ignore[misc] controller_id, fem_id, port_id, **kwargs ) elif port_type == "analog_input": - ports[port_id] = LFFEMAnalogInputPort( + ports[port_id] = LFFEMAnalogInputPort( # type: ignore[misc] controller_id, fem_id, port_id, **kwargs ) elif port_type == "mw_output": @@ -229,7 +231,7 @@ def _get_port( kwargs["upconverter_frequency"] = 5e9 if "band" not in kwargs: kwargs["band"] = 1 - ports[port_id] = MWFEMAnalogOutputPort( + ports[port_id] = MWFEMAnalogOutputPort( # type: ignore[misc] controller_id, fem_id, port_id, **kwargs ) elif port_type == "mw_input": @@ -239,14 +241,14 @@ def _get_port( kwargs["band"] = 1 if "downconverter_frequency" not in kwargs: kwargs["downconverter_frequency"] = 5e9 - ports[port_id] = MWFEMAnalogInputPort( + ports[port_id] = MWFEMAnalogInputPort( # type: ignore[misc] controller_id, fem_id, port_id, **kwargs, ) elif port_type == "digital_output": - ports[port_id] = FEMDigitalOutputPort( + ports[port_id] = FEMDigitalOutputPort( # type: ignore[misc] controller_id, fem_id, port_id, **kwargs ) @@ -266,13 +268,16 @@ def reference_to_port( try: elems = port_reference.split("/") - port_type, controller_id, fem_id, port_id = elems[-4:] + port_type_str, controller_id_str, fem_id_str, port_id_str = elems[-4:] - port_type = port_type[:-1] - if controller_id.isdigit(): - controller_id = int(controller_id) - fem_id = int(fem_id) - port_id = int(port_id) + port_type = port_type_str[:-1] + controller_id: Union[str, int] = ( + int(controller_id_str) + if controller_id_str.isdigit() + else controller_id_str + ) + fem_id: int = int(fem_id_str) + port_id: int = int(port_id_str) except Exception as e: raise ValueError( f"Unable to parse port reference for OPX1000 FEM: {port_reference}" diff --git a/quam/components/pulses.py b/quam/components/pulses.py index 24e7c8fe..9bdad40e 100644 --- a/quam/components/pulses.py +++ b/quam/components/pulses.py @@ -2,7 +2,6 @@ import warnings from abc import ABC, abstractmethod from collections.abc import Iterable -from dataclasses import field from typing import Any, ClassVar, Dict, List, Optional, Sequence, Tuple, Union import numpy as np @@ -85,9 +84,9 @@ class Pulse(QuamComponent): operation: ClassVar[str] = "control" length: int - id: str = None + id: Optional[str] = None - digital_marker: Union[str, List[Tuple[int, int]]] = None + digital_marker: Optional[Union[str, List[Tuple[int, int]]]] = None @property def channel(self): @@ -147,7 +146,7 @@ def _get_sampling_rate(self) -> float: def calculate_waveform( self, - ) -> Union[float, complex, Sequence[float], Sequence[complex]]: + ) -> Optional[Union[float, complex, Sequence[float], Sequence[complex]]]: """Calculate the waveform of the pulse. This function calls `Pulse.waveform_function`, which should generally be @@ -168,21 +167,25 @@ def calculate_waveform( # Optionally convert IQ waveforms to complex waveform if isinstance(waveform, tuple) and len(waveform) == 2: if isinstance(waveform[0], (list, np.ndarray)): - waveform = np.array(waveform[0]) + 1.0j * np.array(waveform[1]) - else: - waveform = waveform[0] + 1.0j * waveform[1] + waveform = np.array(waveform[0]) + 1.0j * np.array(waveform[1]) # type: ignore[assignment] + elif isinstance(waveform[0], (int, float)) and isinstance( + waveform[1], (int, float) + ): + waveform = complex(waveform[0], waveform[1]) return waveform def waveform_function( self, - ) -> Union[ - float, - complex, - Sequence[float], - Sequence[complex], - Tuple[float, float], - Tuple[Sequence[float], Sequence[float]], + ) -> Optional[ + Union[ + float, + complex, + Sequence[float], + Sequence[complex], + Tuple[float, float], + Tuple[Sequence[float], Sequence[float]], + ] ]: """Function that returns the waveform of the pulse. @@ -192,7 +195,7 @@ def waveform_function( This function is called from `Pulse.calculate_waveform` Returns: - The waveform of the pulse. Can be one of the following: + The waveform of the pulse, or None for digital-only pulses. Can be one of: - a single float for a constant single-channel waveform, - a single complex number for a constant IQ waveform, - a sequence of floats for an arbitrary single-channel waveform, @@ -200,16 +203,16 @@ def waveform_function( - a tuple of floats for a constant IQ waveform, - a tuple of sequences for an arbitrary IQ waveform """ - ... + return None def play( self, amplitude_scale: Optional[Union[ScalarFloat, Sequence[ScalarFloat]]] = None, - duration: ScalarInt = None, - condition: ScalarBool = None, - chirp: ChirpType = None, - truncate: ScalarInt = None, - timestamp_stream: StreamType = None, + duration: Optional[ScalarInt] = None, + condition: Optional[ScalarBool] = None, + chirp: Optional[ChirpType] = None, + truncate: Optional[ScalarInt] = None, + timestamp_stream: Optional[StreamType] = None, continue_chirp: bool = False, target: str = "", validate: bool = True, @@ -256,6 +259,7 @@ def play( ValueError: If the pulse is not attached to a channel. KeyError: If the pulse is not registered in the channel's operations. """ + name: Union[str, int] if self.id is not None: name = self.id elif self.parent is not None: @@ -287,7 +291,7 @@ def _config_add_pulse(self, config: Dict[str, Any]): assert self.operation in ["control", "measurement"] assert isinstance(self.length, int) - pulse_config = config["pulses"][self.pulse_name] = { + config["pulses"][self.pulse_name] = { "operation": self.operation, "length": self.length, } @@ -392,7 +396,7 @@ def _config_add_digital_markers(self, config): def apply_to_config(self, config: dict) -> None: """Adds this pulse, waveform, and digital marker to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ if self.channel is None: @@ -422,8 +426,8 @@ class BaseReadoutPulse(Pulse, ABC): digital_marker: str = "ON" # TODO Understand why the thresholds were added. - threshold: float = None - rus_exit_threshold: float = None + threshold: Optional[float] = None + rus_exit_threshold: Optional[float] = None _weight_labels: ClassVar[List[str]] = ["iw1", "iw2", "iw3"] @@ -468,7 +472,7 @@ def _config_add_integration_weights(self, config: dict): def apply_to_config(self, config: dict) -> None: """Adds this readout pulse to the QUA configuration. - See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] + See [`QuamComponent.apply_to_config`][quam.core.quam_classes.QuamComponent.apply_to_config] # noqa: E501 for details. """ super().apply_to_config(config) @@ -496,7 +500,7 @@ class ReadoutPulse(BaseReadoutPulse, ABC): integration weights in radians. """ - integration_weights: Union[List[float], List[Tuple[float, int]]] = ( + integration_weights: Union[List[float], List[Tuple[float, int]], str] = ( "#./default_integration_weights" ) integration_weights_angle: float = 0 @@ -505,7 +509,7 @@ class ReadoutPulse(BaseReadoutPulse, ABC): def default_integration_weights(self) -> List[Tuple[float, int]]: return [(1, self.length)] - def integration_weights_function(self) -> List[Tuple[Union[complex, float], int]]: + def integration_weights_function(self) -> Dict[str, List[Tuple[float, int]]]: phase = np.exp(1j * self.integration_weights_angle) if isinstance(self.integration_weights[0], float): @@ -539,9 +543,9 @@ class WaveformPulse(Pulse): waveform_Q: Optional[List[float]] = None # Length is derived from the waveform_I length, but still needs to be declared # to satisfy the dataclass, but we'll override its behavior - length: Optional[int] = None # pyright: ignore + length: Optional[int] = None # type: ignore[assignment] # pyright: ignore - @property + @property # type: ignore[override, no-redef] def length(self): # noqa: 811 if not isinstance(self.waveform_I, Iterable): return None @@ -561,6 +565,7 @@ def to_dict( self, follow_references: bool = False, include_defaults: bool = True ) -> Dict[str, Any]: d = super().to_dict(follow_references, include_defaults) + assert isinstance(d, dict) d.pop("length") return d @@ -709,7 +714,7 @@ class SquarePulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None def waveform_function(self): waveform = self.amplitude @@ -775,7 +780,7 @@ class GaussianPulse(Pulse): amplitude: float length: int sigma: float - axis_angle: float = None + axis_angle: Optional[float] = None subtracted: bool = True def __post_init__(self) -> None: @@ -821,7 +826,7 @@ class FlatTopGaussianPulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int def __post_init__(self) -> None: @@ -880,17 +885,17 @@ class _FlatTopGaussianPulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int smoothing_length: int = 0 post_zero_padding_length: int = 0 - length: int = "#./inferred_total_length" + length: Union[int, str] = "#./inferred_total_length" # type: ignore[assignment] @property def inferred_total_length(self) -> int: warnings.warn( - "_FlatTopGaussianPulse is deprecated and will be removed in a future release. " - "Please use the version in qualang-tools instead.", + "_FlatTopGaussianPulse is deprecated and will be removed in a future " + "release. Please use the version in qualang-tools instead.", DeprecationWarning, stacklevel=2, ) @@ -942,7 +947,7 @@ class FlatTopBlackmanPulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int def __post_init__(self) -> None: @@ -989,7 +994,7 @@ class BlackmanIntegralPulse(Pulse): # amplitude: float v_start: float v_end: float - axis_angle: float = None + axis_angle: Optional[float] = None def __post_init__(self) -> None: warnings.warn( @@ -1025,7 +1030,7 @@ class FlatTopCosinePulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int = 0 def __post_init__(self) -> None: @@ -1070,7 +1075,7 @@ class FlatTopTanhPulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int = 0 def __post_init__(self) -> None: @@ -1138,17 +1143,17 @@ class _CosineBipolarPulse(Pulse): """ amplitude: float - axis_angle: float = None + axis_angle: Optional[float] = None flat_length: int smoothing_length: int = 0 post_zero_padding_length: int = 0 - length: int = "#./inferred_total_length" + length: Union[int, str] = "#./inferred_total_length" # type: ignore[assignment] @property def inferred_total_length(self) -> int: warnings.warn( - "_CosineBipolarPulse is deprecated and will be removed in a future release. " - "Please use the version in qualang-tools instead.", + "_CosineBipolarPulse is deprecated and will be removed in a future " + "release. Please use the version in qualang-tools instead.", DeprecationWarning, stacklevel=2, ) diff --git a/quam/components/quantum_components/qubit.py b/quam/components/quantum_components/qubit.py index b1e62aab..e34af0ec 100644 --- a/quam/components/quantum_components/qubit.py +++ b/quam/components/quantum_components/qubit.py @@ -1,16 +1,14 @@ from collections import UserDict from collections.abc import Iterable -from typing import Dict, List, Optional, Union, TYPE_CHECKING, Any +from typing import Dict, Optional, Union, TYPE_CHECKING, Any from dataclasses import field -from qm import qua from qm.qua import align from quam.components.channels import Channel from quam.components.pulses import Pulse from quam.components.quantum_components import QuantumComponent from quam.core import quam_dataclass -from quam.utils import string_reference as str_ref if TYPE_CHECKING: from quam.components.macro import QubitMacro @@ -26,7 +24,9 @@ @quam_dataclass class Qubit(QuantumComponent): id: Union[str, int] = "#./inferred_id" - macros: Dict[str, MacroType] = field(default_factory=dict) + macros: Dict[str, MacroType] = field( # type: ignore[assignment] + default_factory=dict + ) @property def name(self) -> str: diff --git a/quam/components/quantum_components/qubit_pair.py b/quam/components/quantum_components/qubit_pair.py index 26aa8110..e46dcd92 100644 --- a/quam/components/quantum_components/qubit_pair.py +++ b/quam/components/quantum_components/qubit_pair.py @@ -18,7 +18,9 @@ class QubitPair(QuantumComponent): id: str = "#./name" qubit_control: Qubit qubit_target: Qubit - macros: Dict[str, MacroType] = field(default_factory=dict) + macros: Dict[str, MacroType] = field( # type: ignore[assignment] + default_factory=dict + ) @property def name(self) -> str: diff --git a/quam/config/cli/config.py b/quam/config/cli/config.py index cf6d90ce..3dbeae36 100644 --- a/quam/config/cli/config.py +++ b/quam/config/cli/config.py @@ -100,7 +100,7 @@ def config_command( common_config, quam_top_l.quam, QUAM_CONFIG_KEY, - quam_before_write_cb, + quam_before_write_cb, # type: ignore[arg-type] confirm=not auto_accept, ) except Exit: diff --git a/quam/config/cli/migrations/v1_v2.py b/quam/config/cli/migrations/v1_v2.py index 33a7cd28..dbd34e9e 100644 --- a/quam/config/cli/migrations/v1_v2.py +++ b/quam/config/cli/migrations/v1_v2.py @@ -2,7 +2,7 @@ from qualibrate_config.qulibrate_types import RawConfigType -class Migrate(MigrateBase): +class Migrate(MigrateBase): # type: ignore[misc, valid-type] from_version: int = 1 to_version: int = 2 diff --git a/quam/config/cli/migrations/v2_v3.py b/quam/config/cli/migrations/v2_v3.py index 7f9b8581..0dc32a5f 100644 --- a/quam/config/cli/migrations/v2_v3.py +++ b/quam/config/cli/migrations/v2_v3.py @@ -2,7 +2,7 @@ from qualibrate_config.qulibrate_types import RawConfigType -class Migrate(MigrateBase): +class Migrate(MigrateBase): # type: ignore[misc, valid-type] from_version: int = 2 to_version: int = 3 diff --git a/quam/config/resolvers.py b/quam/config/resolvers.py index a5690944..4dc06fe9 100644 --- a/quam/config/resolvers.py +++ b/quam/config/resolvers.py @@ -13,7 +13,6 @@ from quam.config.models import QuamConfig, QuamTopLevelConfig from quam.config.validators import ( quam_version_validator, - InvalidQuamConfigVersion, InvalidQuamConfigVersionError, GreaterThanSupportedQuamConfigVersionError, ) diff --git a/quam/core/__init__.py b/quam/core/__init__.py index 3764040e..20ada9b8 100644 --- a/quam/core/__init__.py +++ b/quam/core/__init__.py @@ -1,10 +1,2 @@ from .quam_classes import * from .operation.operations_registry import OperationsRegistry - -# Exec statement needed to trick Pycharm type checker into recognizing it as a dataclass -# This will no longer be necessary once we drop support for Python 3.9 -# We also do this at the end of quam_classes.py, but PyCharm also requires it here -_quam_dataclass = quam_dataclass -from dataclasses import dataclass as quam_dataclass - -exec("quam_dataclass = _quam_dataclass") diff --git a/quam/core/deprecations.py b/quam/core/deprecations.py index 33b9796a..5cc0307b 100644 --- a/quam/core/deprecations.py +++ b/quam/core/deprecations.py @@ -1,15 +1,17 @@ import warnings -from abc import abstractclassmethod +from abc import ABC, abstractmethod instantiation_deprecations = [] -class InstantiationDeprecationRule: - @abstractclassmethod +class InstantiationDeprecationRule(ABC): + @classmethod + @abstractmethod def match(cls, quam_class, contents): raise NotImplementedError - @abstractclassmethod + @classmethod + @abstractmethod def apply(cls, quam_class, contents): raise NotImplementedError diff --git a/quam/core/macro/quam_macro.py b/quam/core/macro/quam_macro.py index 9c24552b..9611376e 100644 --- a/quam/core/macro/quam_macro.py +++ b/quam/core/macro/quam_macro.py @@ -1,7 +1,6 @@ from abc import ABC from typing import Optional, Union from quam.core.quam_classes import quam_dataclass, QuamComponent -from quam.utils import string_reference as str_ref from quam.core.macro.base_macro import BaseMacro __all__ = ["QuamMacro"] @@ -11,7 +10,7 @@ class QuamMacro(QuamComponent, BaseMacro, ABC): id: str = "#./inferred_id" fidelity: Optional[float] = None - duration: Optional[float] = "#./inferred_duration" + duration: Union[Optional[float], str] = "#./inferred_duration" @property def inferred_duration(self) -> Optional[float]: diff --git a/quam/core/operation/function_properties.py b/quam/core/operation/function_properties.py index 41ccc1f1..72454841 100644 --- a/quam/core/operation/function_properties.py +++ b/quam/core/operation/function_properties.py @@ -1,4 +1,4 @@ -from typing import Any, Callable, Optional, Type, get_origin, get_args, TypeVar +from typing import Any, Callable, Optional, Type import inspect from typing import get_type_hints from dataclasses import dataclass, field @@ -9,9 +9,6 @@ __all__ = ["FunctionProperties"] -QC = TypeVar("QC", bound=QuantumComponent) - - @dataclass class FunctionProperties: """ @@ -29,7 +26,7 @@ class FunctionProperties: """ quantum_component_name: str - quantum_component_type: Type[QC] + quantum_component_type: Type[QuantumComponent] name: str = "" required_args: list[str] = field(default_factory=list) optional_args: dict[str, Any] = field(default_factory=dict) diff --git a/quam/core/operation/operation.py b/quam/core/operation/operation.py index 3dccb9e2..198269d8 100644 --- a/quam/core/operation/operation.py +++ b/quam/core/operation/operation.py @@ -1,4 +1,4 @@ -from typing import Callable, Optional, Any +from typing import Callable from quam.core.operation.function_properties import FunctionProperties from quam.components import QuantumComponent @@ -11,7 +11,8 @@ def __init__(self, func: Callable): """ Initialize a quantum operation. - This is typically used implicitly from the decorator @operations_registry.register_operation. + This is typically used implicitly from the decorator + @operations_registry.register_operation. Args: func: The function implementing the operation diff --git a/quam/core/operation/operations_registry.py b/quam/core/operation/operations_registry.py index 109685d0..27c7b691 100644 --- a/quam/core/operation/operations_registry.py +++ b/quam/core/operation/operations_registry.py @@ -1,6 +1,6 @@ from collections import UserDict import functools -from typing import Callable, Optional, TypeVar, Any +from typing import Callable, TypeVar, Any from quam.core.operation import Operation @@ -12,7 +12,7 @@ class OperationsRegistry(UserDict): """A registry to store and manage operations.""" - def register_operation(self, func: Optional[T]) -> T: + def register_operation(self, func: T) -> T: """ Register a function as an operation. @@ -31,7 +31,7 @@ def register_operation(self, func: Optional[T]) -> T: # return functools.partial(self.register_operation) operation = Operation(func) - operation = functools.update_wrapper(operation, func) + functools.update_wrapper(operation, func) self[func.__name__] = operation diff --git a/quam/core/quam_classes.py b/quam/core/quam_classes.py index f6edf22e..ac2c0699 100644 --- a/quam/core/quam_classes.py +++ b/quam/core/quam_classes.py @@ -1,12 +1,9 @@ from __future__ import annotations from collections.abc import Iterable -from collections import UserList -import sys import warnings from pathlib import Path from copy import deepcopy from typing import ( - Iterator, Union, Generator, ClassVar, @@ -15,12 +12,18 @@ Dict, Sequence, TypeVar, + cast, get_type_hints, get_origin, get_args, Optional, ) from functools import partial + +try: + from typing import dataclass_transform +except ImportError: + from typing_extensions import dataclass_transform # type: ignore[no-redef] from dataclasses import dataclass, fields, is_dataclass, MISSING from collections import UserDict, UserList @@ -84,7 +87,7 @@ def convert_dict_and_list(value, cls_or_obj=None, attr=None): if isinstance(value, dict): value_annotation = _get_value_annotation(cls_or_obj=cls_or_obj, attr=attr) return QuamDict(value, value_annotation=value_annotation) - elif type(value) == list: + elif type(value) is list: value_annotation = _get_value_annotation(cls_or_obj=cls_or_obj, attr=attr) return QuamList(value, value_annotation=value_annotation) else: @@ -151,44 +154,18 @@ def sort_quam_components( return sorted_components -def _quam_dataclass(cls=None, **kwargs): - """Dataclass for QUAM classes. - - This class is used as a patch to maintain compatibility with Python 3.9, as - these do not support the dataclass argument `kw_only`. This argument is needed to - ensure inheritance of parent dataclasses is allowed. +@dataclass_transform(eq_default=False, kw_only_default=True) +def quam_dataclass(cls=None, **kwargs): + """Dataclass decorator for QUAM classes. - Args: - - cls: The QUAM class to decorate. - - kwargs: The arguments to pass to the dataclass decorator. - By default, kw_only=True and eq=False are passed, though they can be overwritten. - Notes: - - This custom dataclass is no longer necessary once Python 3.9 support is dropped - - The actual custom dataclass is `quam_dataclass` (without the underscore). This - function is only used to trick type checkers into recognizing it as a dataclass. - - From Python 3.10 onwards, this customized dataclass is no longer needed, as then - the following two decorators are equivalent: - - @quam_dataclass - - @dataclass(eq=False, kw_only=True) + Equivalent to @dataclass(eq=False, kw_only=True). Both defaults can be overridden + by passing the corresponding keyword argument, e.g. @quam_dataclass(kw_only=False). """ if cls is None: - return partial(_quam_dataclass, **kwargs) - + return partial(quam_dataclass, **kwargs) kwargs.setdefault("kw_only", True) kwargs.setdefault("eq", False) - - if sys.version_info.minor > 9: - return dataclass(cls, **kwargs) - - from quam.utils.dataclass import _quam_patched_dataclass - - return _quam_patched_dataclass(cls, **kwargs) - - -# Exec statement is needed to trick type checkers into recognizing it as a dataclass -# This will no longer be necessary once we drop support for Python 3.9 -quam_dataclass = dataclass -exec("quam_dataclass = _quam_dataclass") + return dataclass(cls, **kwargs) class ParentDescriptor: @@ -245,9 +222,9 @@ class QuamBase(ReferenceClass): The subclasses should be dataclasses. """ - parent: ClassVar["QuamBase"] = ParentDescriptor() + parent: Optional["QuamBase"] = ParentDescriptor() # type: ignore[assignment] _last_instantiated_root: ClassVar[Optional["QuamRoot"]] = None - config_settings: ClassVar[Dict[str, Any]] = None + config_settings: ClassVar[Optional[Dict[str, Any]]] = None _MAX_REFERENCE_DEPTH: ClassVar[int] = 10 def __init__(self): @@ -299,7 +276,7 @@ def get_root(self) -> Optional[QuamRoot]: return self._last_instantiated_root return None - def get_attr_name(self, attr_val: Any) -> str: + def get_attr_name(self, attr_val: Any) -> Union[str, int]: """Get the name of an attribute that matches the value. Args: @@ -394,7 +371,7 @@ def _val_matches_attr_annotation(cls, attr: str, val: Any) -> bool: return isinstance(val, (dict, QuamDict)) elif required_type == list or get_origin(required_type) == list: return isinstance(val, (list, QuamList)) - return type(val) == required_type + return type(val) is required_type def get_reference( self, @@ -537,7 +514,7 @@ def get_attrs( def to_dict( self, follow_references: bool = False, include_defaults: bool = True - ) -> Dict[str, Any]: + ) -> Union[Dict[str, Any], List[Any]]: """Convert this object to a dictionary. Args: @@ -557,7 +534,7 @@ def to_dict( attrs = self.get_attrs( follow_references=follow_references, include_defaults=include_defaults ) - quam_dict = {} + quam_dict: Dict[str, Any] = {} for attr, val in attrs.items(): if isinstance(val, QuamBase): quam_dict[attr] = val.to_dict( @@ -573,12 +550,12 @@ def to_dict( return quam_dict def iterate_components( - self, skip_elems: Optional[Sequence["QuamBase"]] = None + self, skip_elems: Optional[List["QuamBase"]] = None ) -> Generator["QuamBase", None, None]: """Iterate over all QuamBase objects in this object, including nested objects. Args: - skip_elems: A sequence of QuamBase objects to skip. + skip_elems: A list of QuamBase objects to skip. This is used to prevent infinite loops when iterating over nested objects. @@ -806,13 +783,10 @@ def set_at_reference(self, attr: str, value: Any, allow_non_reference: bool = Tr target_obj, target_attr = self._follow_reference_chain(self, attr) # Use __setitem__ for dict/list types, otherwise use setattr - if isinstance(target_obj, (dict, list, UserDict, UserList, QuamDict, QuamList)): + if isinstance(target_obj, (list, UserList, QuamList)): # Convert string index to int for list types - if ( - isinstance(target_obj, (list, UserList, QuamList)) - and target_attr.isdigit() - ): - target_attr = int(target_attr) + target_obj[int(target_attr) if target_attr.isdigit() else target_attr] = value # type: ignore[index] + elif isinstance(target_obj, (dict, UserDict, QuamDict)): target_obj[target_attr] = value else: setattr(target_obj, target_attr, value) @@ -854,8 +828,15 @@ def get_serialiser(cls) -> AbstractSerialiser: return JSONSerialiser() def get_reference( - self, attr: Optional[str] = None, relative_path: Optional[str] = None + self, + attr: Optional[str] = None, + relative_path: Optional[str] = None, + follow_chain: bool = False, ) -> Optional[str]: + # `follow_chain` is accepted for signature compatibility with QuamBase; + # a root has no chain to follow. + if follow_chain: + raise NotImplementedError("follow_chain is not supported on QuamRoot") if attr is not None: return f"#/{attr}" @@ -931,11 +912,14 @@ def load( contents, _ = serialiser.load(filepath_or_dict) try: - return instantiate_quam_class( - quam_class=cls, - contents=contents, - fix_attrs=fix_attrs, - validate_type=validate_type, + return cast( + QuamRootType, + instantiate_quam_class( + quam_class=cls, + contents=contents, + fix_attrs=fix_attrs, + validate_type=validate_type, + ), ) except (AttributeError, TypeError, ValueError) as e: if not isinstance(filepath_or_dict, dict) and filepath_or_dict is not None: @@ -958,7 +942,7 @@ def generate_config(self) -> DictQuaConfig: """ qua_config = deepcopy(qua_config_template) - quam_components = list(self.iterate_components()) + quam_components = cast(List["QuamComponent"], list(self.iterate_components())) sorted_components = sort_quam_components(quam_components) for quam_component in sorted_components: @@ -966,7 +950,7 @@ def generate_config(self) -> DictQuaConfig: generate_config_final_actions(qua_config) - return qua_config + return cast(DictQuaConfig, qua_config) class QuamComponent(QuamBase): @@ -1026,9 +1010,9 @@ class QuamDict(UserDict, QuamBase): it can be used as a normal dictionary, but it is not a subclass of `dict`. """ - _value_annotation: ClassVar[type] = None + _value_annotation: ClassVar[Optional[type]] = None - def __init__(self, dict=None, /, value_annotation: type = None, **kwargs): + def __init__(self, dict=None, /, value_annotation: Optional[type] = None, **kwargs): self.__dict__["data"] = {} self.__dict__["_value_annotation"] = value_annotation self.__dict__["_initialized"] = True @@ -1127,7 +1111,7 @@ def get_attr_name(self, attr_val: Any) -> Union[str, int]: f"obj: {self}" ) - def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: + def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: # type: ignore[override] """Check whether the type of an attribute matches the annotation. Called by [`QuamDict.to_dict`][quam.core.quam_classes.QuamDict.to_dict] to @@ -1145,7 +1129,7 @@ def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: return True if self._value_annotation is None: return False - return type(val) == self._value_annotation + return type(val) is self._value_annotation def _attr_val_is_default(self, attr: str, val: Any): """Check whether the value of an attribute is the default value. @@ -1175,12 +1159,12 @@ def get_raw_value(self, attr: str) -> Any: ) from e def iterate_components( - self, skip_elems: Sequence[QuamBase] = None + self, skip_elems: Optional[List[QuamBase]] = None ) -> Generator["QuamBase", None, None]: """Iterate over all QuamBase objects in this object, including nested objects. Args: - skip_elems: A sequence of QuamBase objects to skip. + skip_elems: A list of QuamBase objects to skip. This is used to prevent infinite loops when iterating over nested objects. @@ -1213,9 +1197,12 @@ def to_dict( Returns: A dictionary representation of the object. """ - quam_dict = super().to_dict( - follow_references=follow_references, - include_defaults=include_defaults, + quam_dict = cast( + Dict[str, Any], + super().to_dict( + follow_references=follow_references, + include_defaults=include_defaults, + ), ) # Remove __class__ from the dictionary as it's the default for a dict @@ -1245,9 +1232,9 @@ class QuamList(UserList, QuamBase): it can be used as a normal list, but it is not a subclass of `list`. """ - _value_annotation: ClassVar[type] = None + _value_annotation: Optional[type] = None - def __init__(self, *args, value_annotation: type = None): + def __init__(self, *args, value_annotation: Optional[type] = None): self._value_annotation = value_annotation # We manually add elements using extend instead of passing to super() @@ -1323,7 +1310,7 @@ def insert(self, i: int, item: Any) -> None: return super().insert(i, converted_item) - def extend(self, iterable: Iterator) -> None: + def extend(self, iterable: Iterable) -> None: converted_iterable = [convert_dict_and_list(elem) for elem in iterable] for converted_item in converted_iterable: if isinstance(converted_item, QuamBase): @@ -1332,7 +1319,7 @@ def extend(self, iterable: Iterator) -> None: return super().extend(converted_iterable) # Quam methods - def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: + def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: # type: ignore[override] """Check whether the type of an attribute matches the annotation. Called by QuamList.to_dict to determine whether to add the __class__ key. @@ -1342,7 +1329,7 @@ def _val_matches_attr_annotation(self, attr: str, val: Any) -> bool: return True if self._value_annotation is None: return False - return type(val) == self._value_annotation + return type(val) is self._value_annotation def get_attr_name(self, attr_val: Any) -> str: for k, elem in enumerate(self.data): @@ -1390,7 +1377,7 @@ def to_dict( return quam_list def iterate_components( - self, skip_elems: List[QuamBase] = None + self, skip_elems: Optional[List[QuamBase]] = None ) -> Generator["QuamBase", None, None]: """Iterate over all QuamBase objects in this object, including nested objects. diff --git a/quam/core/quam_instantiation.py b/quam/core/quam_instantiation.py index 6612781d..0b0f2102 100644 --- a/quam/core/quam_instantiation.py +++ b/quam/core/quam_instantiation.py @@ -53,8 +53,6 @@ def instantiate_attrs_from_dict( Returns: A dictionary with the instantiated attributes. """ - from quam.core import QuamComponent # noqa: F811 - if typing.get_origin(required_type) == dict: required_subtype = typing.get_args(required_type)[1] else: @@ -105,8 +103,6 @@ def instantiate_attrs_from_list( Returns: A list with the instantiated attributes. """ - from quam.core import QuamComponent # noqa: F811 - if typing.get_origin(required_type) == list: required_subtype = typing.get_args(required_type)[0] else: @@ -222,7 +218,8 @@ def instantiate_attr( continue else: raise TypeError( - f"Could not instantiate {str_repr} with any of the types in {expected_type}" + f"Could not instantiate {str_repr} with any of the types in " + f"{expected_type}" ) elif ( isinstance(expected_type, list) @@ -290,12 +287,17 @@ class definition. If False, any attribute can be set. validate_type: Whether to validate the type of the attributes. A TypeError is raised if an attribute has the wrong type. str_repr: A string representation of the object, used for error messages. - quam_class: The QuamBase class being instantiated, used for richer error messages. + quam_class: The QuamBase class being instantiated, used for richer + error messages. Returns: A dictionary where each element has been instantiated if it is a QuamComponent """ - instantiated_attrs = {"required": {}, "optional": {}, "extra": {}} + instantiated_attrs: Dict[str, Dict[str, Any]] = { + "required": {}, + "optional": {}, + "extra": {}, + } for attr_name, attr_val in contents.items(): if attr_name in ("__class__", "__package_versions__"): continue @@ -402,7 +404,7 @@ def instantiate_quam_class( if not str_repr: str_repr = quam_class.__name__ - # str_repr = f"{str_repr}.{quam_class.__name__}" if str_repr else quam_class.__name__ + # str_repr = f"{str_repr}.{quam_class.__name__}" if str_repr else quam_class.__name__ # noqa: E501 if "__class__" in contents: try: diff --git a/quam/examples/superconducting_qubits/components.py b/quam/examples/superconducting_qubits/components.py index e22bb08e..0099a80d 100644 --- a/quam/examples/superconducting_qubits/components.py +++ b/quam/examples/superconducting_qubits/components.py @@ -1,5 +1,5 @@ from dataclasses import field -from typing import Dict, List, Union +from typing import Dict, Optional, Union from quam import QuamComponent from quam.components.channels import IQChannel, SingleChannel, InOutIQChannel @@ -14,10 +14,10 @@ class Transmon(QuamComponent): id: Union[int, str] - xy: IQChannel = None - z: SingleChannel = None + xy: Optional[IQChannel] = None + z: Optional[SingleChannel] = None - resonator: InOutIQChannel = None + resonator: Optional[InOutIQChannel] = None @property def name(self): diff --git a/quam/examples/superconducting_qubits/generate_superconducting_quam.py b/quam/examples/superconducting_qubits/generate_superconducting_quam.py index 73471d99..422cde7f 100644 --- a/quam/examples/superconducting_qubits/generate_superconducting_quam.py +++ b/quam/examples/superconducting_qubits/generate_superconducting_quam.py @@ -1,8 +1,9 @@ from pathlib import Path import json -from quam.components import * +from quam.components import * # noqa: F401, F403 from quam.components.channels import IQChannel, InOutIQChannel, SingleChannel +from quam.components.hardware import FrequencyConverter, Mixer, LocalOscillator from quam.examples.superconducting_qubits.components import Transmon, Quam from quam.core import QuamRoot @@ -36,29 +37,36 @@ def create_quam_superconducting_referenced(num_qubits: int) -> QuamRoot: for idx in range(num_qubits): # Create qubit components - transmon = Transmon(id=idx) + transmon = Transmon(id=idx) # type: ignore[call-arg] machine.qubits[transmon.name] = transmon - transmon.xy = IQChannel( - opx_output_I=f"#/wiring/qubits/q{idx}/port_I", - opx_output_Q=f"#/wiring/qubits/q{idx}/port_Q", - frequency_converter_up=FrequencyConverter( + transmon.xy = IQChannel( # type: ignore[call-arg] + opx_output_I=f"#/wiring/qubits/q{idx}/port_I", # type: ignore[arg-type] + opx_output_Q=f"#/wiring/qubits/q{idx}/port_Q", # type: ignore[arg-type] + frequency_converter_up=FrequencyConverter( # type: ignore[call-arg] mixer=Mixer(), - local_oscillator=LocalOscillator(power=10, frequency=6e9), + local_oscillator=LocalOscillator( # type: ignore[call-arg] + power=10, frequency=6e9 + ), ), intermediate_frequency=100e6, ) - transmon.z = SingleChannel(opx_output=f"#/wiring/qubits/q{idx}/port_Z") + transmon.z = SingleChannel( # type: ignore[call-arg] + opx_output=f"#/wiring/qubits/q{idx}/port_Z" # type: ignore[arg-type] + ) - transmon.resonator = InOutIQChannel( + transmon.resonator = InOutIQChannel( # type: ignore[call-arg] id=idx, - opx_output_I="#/wiring/feedline/opx_output_I", - opx_output_Q="#/wiring/feedline/opx_output_Q", - opx_input_I="#/wiring/feedline/opx_input_I", - opx_input_Q="#/wiring/feedline/opx_input_Q", - frequency_converter_up=FrequencyConverter( - mixer=Mixer(), local_oscillator=LocalOscillator(power=10, frequency=6e9) + opx_output_I="#/wiring/feedline/opx_output_I", # type: ignore[arg-type] + opx_output_Q="#/wiring/feedline/opx_output_Q", # type: ignore[arg-type] + opx_input_I="#/wiring/feedline/opx_input_I", # type: ignore[arg-type] + opx_input_Q="#/wiring/feedline/opx_input_Q", # type: ignore[arg-type] + frequency_converter_up=FrequencyConverter( # type: ignore[call-arg] + mixer=Mixer(), + local_oscillator=LocalOscillator( # type: ignore[call-arg] + power=10, frequency=6e9 + ), ), ) return machine diff --git a/quam/examples/superconducting_qubits/operations.py b/quam/examples/superconducting_qubits/operations.py index 368ad1ad..c06c7f6e 100644 --- a/quam/examples/superconducting_qubits/operations.py +++ b/quam/examples/superconducting_qubits/operations.py @@ -1,5 +1,5 @@ from typing import Tuple -from qm.qua import QuaVariableType +from quam.utils.qua_types import QuaVariableInt from quam.components import Qubit, QubitPair from quam.core import OperationsRegistry @@ -27,7 +27,7 @@ def cz(qubit_pair: QubitPair, **kwargs): @operations_registry.register_operation -def measure(qubit: Qubit, **kwargs) -> QuaVariableType: +def measure(qubit: Qubit, **kwargs) -> QuaVariableInt: # type: ignore[empty-body] pass diff --git a/quam/serialisation/json.py b/quam/serialisation/json.py index 7dfdfdc6..93f052a2 100644 --- a/quam/serialisation/json.py +++ b/quam/serialisation/json.py @@ -325,7 +325,7 @@ def _save_split_content( default_filepath = folder / self.default_filename self._save_dict_to_json(remaining_contents, default_filepath) - def save( + def save( # type: ignore[override] self, quam_obj: QuamRoot, path: Optional[Union[Path, str]] = None, @@ -373,6 +373,7 @@ def save( # Get the dictionary representation of the object contents_dict = quam_obj.to_dict(include_defaults=current_include_defaults) + assert isinstance(contents_dict, dict), "QuamRoot.to_dict() must return a dict" package_versions = collect_package_versions(contents_dict) if package_versions: @@ -380,7 +381,7 @@ def save( # Apply ignore filter directly to the source dictionary before saving # This modification is temporary for the save operation. - effective_contents = contents_dict.copy() + effective_contents: Dict[str, Any] = contents_dict.copy() if ignore: current_content_mapping = current_content_mapping.copy() for key in ignore: @@ -487,7 +488,7 @@ def _load_from_file(self, filepath: Path) -> Tuple[Dict[str, Any], Dict[str, Any f"File {filepath} does not contain a valid JSON dictionary.", ) - metadata = { + metadata: Dict[str, Any] = { "content_mapping": {}, "default_filename": filepath.name, "default_foldername": None, diff --git a/quam/utils/__init__.py b/quam/utils/__init__.py index 5ecf684d..ab4638c7 100644 --- a/quam/utils/__init__.py +++ b/quam/utils/__init__.py @@ -5,6 +5,7 @@ from .general import * from . import string_reference from .config import * +from . import dataclass, general, pulse, reference_class, type_checking, config __all__ = [ *dataclass.__all__, diff --git a/quam/utils/dataclass.py b/quam/utils/dataclass.py index 12b5a2ad..764951e7 100644 --- a/quam/utils/dataclass.py +++ b/quam/utils/dataclass.py @@ -4,6 +4,7 @@ import sys import warnings from typing import Any, Dict, Union, ClassVar, get_type_hints +from dataclasses import Field __all__ = ["patch_dataclass", "get_dataclass_attr_annotations"] @@ -78,7 +79,7 @@ def get_dataclass_attr_annotations( annotated_attrs.pop("_value_annotation", None) annotated_attrs.pop("_initialized", None) - attr_annotations = {"required": {}, "optional": {}} + attr_annotations: Dict[str, Dict[str, type]] = {"required": {}, "optional": {}} for attr, attr_type in annotated_attrs.items(): if getattr(attr_type, "__origin__", None) == ClassVar: continue @@ -87,7 +88,8 @@ def get_dataclass_attr_annotations( if attr not in getattr(cls_or_obj, "__dataclass_fields__", {}): attr_annotations["required"][attr] = attr_type else: - field_attr = cls_or_obj.__dataclass_fields__[attr] + dataclass_fields = getattr(cls_or_obj, "__dataclass_fields__", {}) + field_attr = dataclass_fields[attr] if field_attr.default_factory is not dataclasses.MISSING: attr_annotations["optional"][attr] = attr_type else: @@ -95,7 +97,8 @@ def get_dataclass_attr_annotations( elif hasattr(cls_or_obj, attr): attr_annotations["optional"][attr] = attr_type elif attr in getattr(cls_or_obj, "__dataclass_fields__", {}): - data_field = cls_or_obj.__dataclass_fields__[attr] + dataclass_fields = getattr(cls_or_obj, "__dataclass_fields__", {}) + data_field = dataclass_fields[attr] if data_field.default_factory is not dataclasses.MISSING: attr_annotations["optional"][attr] = attr_type else: @@ -109,7 +112,7 @@ def get_dataclass_attr_annotations( return attr_annotations -def dataclass_field_has_default(field: dataclasses.field) -> bool: +def dataclass_field_has_default(field: Field[Any]) -> bool: """Check if a dataclass field has a default value""" if field.default is not dataclasses.MISSING: return True @@ -216,7 +219,7 @@ def patch_dataclass(module_name): This function should be called at the top of a file, before dataclasses are defined: ``` - patch_dataclass(__name__) # Ensure dataclass "kw_only" also works with python < 3.10 + patch_dataclass(__name__) # Ensure "kw_only" also works with python < 3.10 ``` Prior to Python 3.10, it was not possible for a dataclass to be a subclass of diff --git a/quam/utils/general.py b/quam/utils/general.py index c0402d6d..ce537b23 100644 --- a/quam/utils/general.py +++ b/quam/utils/general.py @@ -10,7 +10,12 @@ from typeguard import TypeCheckError, check_type -__all__ = ["get_full_class_path", "validate_obj_type", "get_class_from_path", "collect_package_versions"] +__all__ = [ + "get_full_class_path", + "validate_obj_type", + "get_class_from_path", + "collect_package_versions", +] def get_full_class_path(cls_or_obj: Union[type, object]) -> str: diff --git a/quam/utils/pulse.py b/quam/utils/pulse.py index f55b698a..44ae9b96 100644 --- a/quam/utils/pulse.py +++ b/quam/utils/pulse.py @@ -56,13 +56,13 @@ def add_amplitude_scale_to_pulse_name( try: check_type(amplitude_scale, Sequence[ScalarFloat]) - return pulse_name * qua.amp(*amplitude_scale) + return pulse_name * qua.amp(*amplitude_scale) # type: ignore[misc, arg-type] except TypeCheckError: pass try: check_type(amplitude_scale, ScalarFloat) - return pulse_name * qua.amp(amplitude_scale) + return pulse_name * qua.amp(amplitude_scale) # type: ignore[arg-type] except TypeCheckError: pass diff --git a/quam/utils/qua_types.py b/quam/utils/qua_types.py index 2df4cb99..c87a84be 100644 --- a/quam/utils/qua_types.py +++ b/quam/utils/qua_types.py @@ -1,3 +1,5 @@ +from typing import TypeAlias + __all__ = [ "ScalarInt", "ScalarFloat", @@ -14,32 +16,31 @@ try: from qm.qua.type_hints import Scalar, QuaScalar, QuaVariable, ChirpType, StreamType - ScalarInt = Scalar[int] - ScalarFloat = Scalar[float] - ScalarBool = Scalar[bool] - QuaScalarInt = QuaScalar[int] + ScalarInt: TypeAlias = Scalar[int] + ScalarFloat: TypeAlias = Scalar[float] + ScalarBool: TypeAlias = Scalar[bool] + QuaScalarInt: TypeAlias = QuaScalar[int] - QuaScalarFloat = QuaScalar[float] - QuaVariable = QuaVariable - QuaVariableBool = QuaVariable[bool] - QuaVariableInt = QuaVariable[int] - QuaVariableFloat = QuaVariable[float] + QuaScalarFloat: TypeAlias = QuaScalar[float] + QuaVariableBool: TypeAlias = QuaVariable[bool] + QuaVariableInt: TypeAlias = QuaVariable[int] + QuaVariableFloat: TypeAlias = QuaVariable[float] except ImportError: - from qm.qua._dsl import ( + from qm.qua._dsl import ( # type: ignore[attr-defined, no-redef] QuaNumberType, QuaVariableType, QuaExpressionType, - ChirpType, - StreamType, + ChirpType, # type: ignore[no-redef] + StreamType, # type: ignore[no-redef] ) - ScalarInt = QuaNumberType - ScalarFloat = QuaNumberType - ScalarBool = QuaExpressionType - QuaScalarInt = QuaNumberType - QuaScalarFloat = QuaNumberType - QuaVariable = QuaVariableType - QuaVariableBool = QuaVariableType - QuaVariableInt = QuaVariableType - QuaVariableFloat = QuaVariableType + ScalarInt: TypeAlias = QuaNumberType # type: ignore[misc, no-redef] + ScalarFloat: TypeAlias = QuaNumberType # type: ignore[misc, no-redef] + ScalarBool: TypeAlias = QuaExpressionType # type: ignore[misc, no-redef] + QuaScalarInt: TypeAlias = QuaNumberType # type: ignore[misc, no-redef] + QuaScalarFloat: TypeAlias = QuaNumberType # type: ignore[misc, no-redef] + QuaVariable: TypeAlias = QuaVariableType # type: ignore[no-redef] + QuaVariableBool: TypeAlias = QuaVariableType # type: ignore[misc, no-redef] + QuaVariableInt: TypeAlias = QuaVariableType # type: ignore[misc, no-redef] + QuaVariableFloat: TypeAlias = QuaVariableType # type: ignore[misc, no-redef] diff --git a/quam/utils/reference_class.py b/quam/utils/reference_class.py index 3c08add5..40955292 100644 --- a/quam/utils/reference_class.py +++ b/quam/utils/reference_class.py @@ -1,4 +1,4 @@ -from typing import Any, ClassVar +from typing import Any import warnings from quam.utils.exceptions import InvalidReferenceError @@ -9,7 +9,7 @@ class ReferenceClass: """Class whose attributes can by references to other attributes""" - _initialized: ClassVar[bool] = False + _initialized: bool = False def __post_init__(self) -> None: """Post init function""" diff --git a/quam/utils/string_reference.py b/quam/utils/string_reference.py index 7bfb2c83..2fa0386f 100644 --- a/quam/utils/string_reference.py +++ b/quam/utils/string_reference.py @@ -43,7 +43,8 @@ def split_next_attribute(string: str, splitter: str = "/") -> Tuple[str, str]: return "", "" if splitter in string: - return tuple(string.split(splitter, 1)) + parts = string.split(splitter, 1) + return parts[0], parts[1] return string, "" @@ -178,7 +179,8 @@ def join_references(base, relative): For a relative base (e.g. '#./a/b'): - We allow accumulating extra '..' if we pop everything (no "true root"). - - We don't remove trailing slashes for relative references (e.g. '#../' remains '#../'). + - We don't remove trailing slashes for relative references + (e.g. '#../' remains '#../'). """ # 1) Disallow if 'relative' starts with "#/" (i.e., another absolute path) if relative.startswith("#/"): diff --git a/tests/components/pulses/test_calculate_waveform.py b/tests/components/pulses/test_calculate_waveform.py new file mode 100644 index 00000000..619b0287 --- /dev/null +++ b/tests/components/pulses/test_calculate_waveform.py @@ -0,0 +1,38 @@ +import numpy as np + +from quam.components.pulses import Pulse +from quam.core import quam_dataclass + + +@quam_dataclass +class FloatTuplePulse(Pulse): + i: float + q: float + + def waveform_function(self): + return (self.i, self.q) + + +@quam_dataclass +class NdarrayTuplePulse(Pulse): + def waveform_function(self): + return (np.array([0.1, 0.2, 0.3, 0.4]), np.array([0.5, 0.6, 0.7, 0.8])) + + +def test_calculate_waveform_float_tuple_returns_complex(): + pulse = FloatTuplePulse(length=16, i=0.1, q=0.2) + result = pulse.calculate_waveform() + assert result == complex(0.1, 0.2) + + +def test_calculate_waveform_float_tuple_type(): + pulse = FloatTuplePulse(length=16, i=0.1, q=0.2) + result = pulse.calculate_waveform() + assert isinstance(result, complex) + + +def test_calculate_waveform_ndarray_tuple_returns_complex_ndarray(): + pulse = NdarrayTuplePulse(length=4) + result = pulse.calculate_waveform() + expected = np.array([0.1, 0.2, 0.3, 0.4]) + 1.0j * np.array([0.5, 0.6, 0.7, 0.8]) + assert np.all(result == expected)