diff --git a/qiskit/qpy/__init__.py b/qiskit/qpy/__init__.py index 60922f3d3ec2..208fa8e44c10 100644 --- a/qiskit/qpy/__init__.py +++ b/qiskit/qpy/__init__.py @@ -17,13 +17,12 @@ .. currentmodule:: qiskit.qpy -QPY is a binary serialization format for :class:`~.QuantumCircuit` and -:class:`~.ScheduleBlock` objects that is designed to be cross-platform, -Python version agnostic, and backwards compatible moving forward. QPY should -be used if you need a mechanism to save or copy between systems a -:class:`~.QuantumCircuit` or :class:`~.ScheduleBlock` that preserves the full -Qiskit object structure (except for custom attributes defined outside of -Qiskit code). This differs from other serialization formats like +QPY is a binary serialization format for :class:`~.QuantumCircuit` +objects that is designed to be cross-platform, Python version agnostic, +and backwards compatible moving forward. QPY should be used if you need +a mechanism to save or copy between systems a :class:`~.QuantumCircuit` +that preserves the full Qiskit object structure (except for custom attributes +defined outside of Qiskit code). This differs from other serialization formats like `OpenQASM `__ (2.0 or 3.0) which has a different abstraction model and can result in a loss of information contained in the original circuit (or is unable to represent some aspects of the @@ -170,6 +169,14 @@ def open(*args): it to QPY setting ``use_symengine=False``. The resulting file can then be loaded by any later version of Qiskit. +.. note:: + + Starting with Qiskit version 2.0.0, which removed the Pulse module from the library, QPY provides + limited support for loading payloads that include pulse data. Loading a ``ScheduleBlock`` payload, + a :class:`.QpyError` exception will be raised. Loading a payload for a circuit that contained pulse + gates, the output circuit will contain custom instructions **without** calibration data attached + for each pulse gate, leaving them undefined. + QPY format version history -------------------------- @@ -902,7 +909,7 @@ def open(*args): --------- Version 7 adds support for :class:`.~Reference` instruction and serialization of -a :class:`.~ScheduleBlock` program while keeping its reference to subroutines:: +a ``ScheduleBlock`` program while keeping its reference to subroutines:: from qiskit import pulse from qiskit import qpy @@ -974,12 +981,12 @@ def open(*args): Version 5 --------- -Version 5 changes from :ref:`qpy_version_4` by adding support for :class:`.~ScheduleBlock` +Version 5 changes from :ref:`qpy_version_4` by adding support for ``ScheduleBlock`` and changing two payloads the INSTRUCTION metadata payload and the CUSTOM_INSTRUCTION block. These now have new fields to better account for :class:`~.ControlledGate` objects in a circuit. In addition, new payload MAP_ITEM is defined to implement the :ref:`qpy_mapping` block. -With the support of :class:`.~ScheduleBlock`, now :class:`~.QuantumCircuit` can be +With the support of ``ScheduleBlock``, now :class:`~.QuantumCircuit` can be serialized together with :attr:`~.QuantumCircuit.calibrations`, or `Pulse Gates `_. In QPY version 5 and above, :ref:`qpy_circuit_calibrations` payload is @@ -996,7 +1003,7 @@ def open(*args): immediately follows the file header block to represent the program type stored in the file. - When ``type==c``, :class:`~.QuantumCircuit` payload follows -- When ``type==s``, :class:`~.ScheduleBlock` payload follows +- When ``type==s``, ``ScheduleBlock`` payload follows .. note:: @@ -1009,12 +1016,10 @@ def open(*args): SCHEDULE_BLOCK ~~~~~~~~~~~~~~ -:class:`~.ScheduleBlock` is first supported in QPY Version 5. This allows +``ScheduleBlock`` is first supported in QPY Version 5. This allows users to save pulse programs in the QPY binary format as follows: -.. plot:: - :include-source: - :nofigs: +.. code-block:: python from qiskit import pulse, qpy @@ -1027,13 +1032,6 @@ def open(*args): with open('schedule.qpy', 'rb') as fd: new_schedule = qpy.load(fd)[0] -.. plot:: - :nofigs: - - # This block is hidden from readers. It's cleanup code. - from pathlib import Path - Path("schedule.qpy").unlink() - Note that circuit and schedule block are serialized and deserialized through the same QPY interface. Input data type is implicitly analyzed and no extra option is required to save the schedule block. @@ -1043,7 +1041,7 @@ def open(*args): SCHEDULE_BLOCK_HEADER ~~~~~~~~~~~~~~~~~~~~~ -:class:`~.ScheduleBlock` block starts with the following header: +``ScheduleBlock`` block starts with the following header: .. code-block:: c @@ -1243,8 +1241,8 @@ def open(*args): and ``num_params`` length of INSTRUCTION_PARAM payload for parameters associated to the custom instruction. The ``type`` indicates the class of pulse program which is either, in principle, -:class:`~.ScheduleBlock` or :class:`~.Schedule`. As of QPY Version 5, -only :class:`~.ScheduleBlock` payload is supported. +``ScheduleBlock`` or :class:`~.Schedule`. As of QPY Version 5, +only ``ScheduleBlock`` payload is supported. Finally, :ref:`qpy_schedule_block` payload is packed for each CALIBRATION_DEF entry. .. _qpy_instruction_v5: diff --git a/qiskit/qpy/binary_io/__init__.py b/qiskit/qpy/binary_io/__init__.py index a5948b7d3f1b..46f0bd0473b7 100644 --- a/qiskit/qpy/binary_io/__init__.py +++ b/qiskit/qpy/binary_io/__init__.py @@ -31,6 +31,5 @@ _read_instruction, ) from .schedules import ( - write_schedule_block, read_schedule_block, ) diff --git a/qiskit/qpy/binary_io/circuits.py b/qiskit/qpy/binary_io/circuits.py index 88622791fa44..2297dbf6d34a 100644 --- a/qiskit/qpy/binary_io/circuits.py +++ b/qiskit/qpy/binary_io/circuits.py @@ -648,8 +648,7 @@ def _read_custom_operations(file_obj, version, vectors): def _read_calibrations(file_obj, version, vectors, metadata_deserializer): - calibrations = {} - + """Consume calibrations data, make the file handle point to the next section""" header = formats.CALIBRATION._make( struct.unpack(formats.CALIBRATION_PACK, file_obj.read(formats.CALIBRATION_SIZE)) ) @@ -658,21 +657,20 @@ def _read_calibrations(file_obj, version, vectors, metadata_deserializer): struct.unpack(formats.CALIBRATION_DEF_PACK, file_obj.read(formats.CALIBRATION_DEF_SIZE)) ) name = file_obj.read(defheader.name_size).decode(common.ENCODE) - qubits = tuple( - struct.unpack("!q", file_obj.read(struct.calcsize("!q")))[0] - for _ in range(defheader.num_qubits) - ) - params = tuple( - value.read_value(file_obj, version, vectors) for _ in range(defheader.num_params) - ) - schedule = schedules.read_schedule_block(file_obj, version, metadata_deserializer) + if name: + warnings.warn( + category=UserWarning, + message="Support for loading pulse gates has been removed in Qiskit 2.0. " + f"If `{name}` is in the circuit it will be left as an opaque instruction.", + ) - if name not in calibrations: - calibrations[name] = {(qubits, params): schedule} - else: - calibrations[name][(qubits, params)] = schedule + for _ in range(defheader.num_qubits): # read qubits info + file_obj.read(struct.calcsize("!q")) + + for _ in range(defheader.num_params): # read params info + value.read_value(file_obj, version, vectors) - return calibrations + schedules.read_schedule_block(file_obj, version, metadata_deserializer) def _dumps_register(register, index_map): @@ -1003,34 +1001,6 @@ def _write_custom_operation( return new_custom_instruction -def _write_calibrations(file_obj, calibrations, metadata_serializer, version): - flatten_dict = {} - for gate, caldef in calibrations.items(): - for (qubits, params), schedule in caldef.items(): - key = (gate, qubits, params) - flatten_dict[key] = schedule - header = struct.pack(formats.CALIBRATION_PACK, len(flatten_dict)) - file_obj.write(header) - for (name, qubits, params), schedule in flatten_dict.items(): - # In principle ScheduleBlock and Schedule can be supported. - # As of version 5 only ScheduleBlock is supported. - name_bytes = name.encode(common.ENCODE) - defheader = struct.pack( - formats.CALIBRATION_DEF_PACK, - len(name_bytes), - len(qubits), - len(params), - type_keys.Program.assign(schedule), - ) - file_obj.write(defheader) - file_obj.write(name_bytes) - for qubit in qubits: - file_obj.write(struct.pack("!q", qubit)) - for param in params: - value.write_value(file_obj, param, version=version) - schedules.write_schedule_block(file_obj, schedule, metadata_serializer, version=version) - - def _write_registers(file_obj, in_circ_regs, full_bits): bitmap = {bit: index for index, bit in enumerate(full_bits)} @@ -1331,8 +1301,11 @@ def write_circuit( file_obj.write(instruction_buffer.getvalue()) instruction_buffer.close() - # Write calibrations - _write_calibrations(file_obj, circuit._calibrations_prop, metadata_serializer, version=version) + # Pulse has been removed in Qiskit 2.0. As long as we keep QPY at version 13, + # we need to write an empty calibrations header since read_circuit expects it + header = struct.pack(formats.CALIBRATION_PACK, 0) + file_obj.write(header) + _write_layout(file_obj, circuit) @@ -1460,11 +1433,9 @@ def read_circuit(file_obj, version, metadata_deserializer=None, use_symengine=Fa standalone_var_indices, ) - # Read calibrations + # Consume calibrations, but don't use them since pulse gates are not supported as of Qiskit 2.0 if version >= 5: - circ._calibrations_prop = _read_calibrations( - file_obj, version, vectors, metadata_deserializer - ) + _read_calibrations(file_obj, version, vectors, metadata_deserializer) for vec_name, (vector, initialized_params) in vectors.items(): if len(initialized_params) != len(vector): diff --git a/qiskit/qpy/binary_io/schedules.py b/qiskit/qpy/binary_io/schedules.py index 4afddb29992a..c2c43af35eec 100644 --- a/qiskit/qpy/binary_io/schedules.py +++ b/qiskit/qpy/binary_io/schedules.py @@ -10,11 +10,15 @@ # copyright notice, and modified files need to carry a notice indicating # that they have been altered from the originals. -"""Read and write schedule and schedule instructions.""" +"""Read schedule and schedule instructions. + +This module is kept post pulse-removal to allow reading legacy +payloads containing pulse gates without breaking the load flow. +The purpose of the `_read` and `_load` methods below is just to advance +the file handle while consuming pulse data.""" import json import struct import zlib -import warnings from io import BytesIO @@ -22,25 +26,17 @@ import symengine as sym from qiskit.exceptions import QiskitError -from qiskit.pulse import library, channels, instructions -from qiskit.pulse.schedule import ScheduleBlock from qiskit.qpy import formats, common, type_keys from qiskit.qpy.binary_io import value from qiskit.qpy.exceptions import QpyError -from qiskit.pulse.configuration import Kernel, Discriminator -from qiskit.utils.deprecate_pulse import ignore_pulse_deprecation_warnings - - -def _read_channel(file_obj, version): - type_key = common.read_type_key(file_obj) - index = value.read_value(file_obj, version, {}) - channel_cls = type_keys.ScheduleChannel.retrieve(type_key) - return channel_cls(index) +def _read_channel(file_obj, version) -> None: + common.read_type_key(file_obj) # read type_key + value.read_value(file_obj, version, {}) # read index -def _read_waveform(file_obj, version): +def _read_waveform(file_obj, version) -> None: header = formats.WAVEFORM._make( struct.unpack( formats.WAVEFORM_PACK, @@ -48,15 +44,8 @@ def _read_waveform(file_obj, version): ) ) samples_raw = file_obj.read(header.data_size) - samples = common.data_from_binary(samples_raw, np.load) - name = value.read_value(file_obj, version, {}) - - return library.Waveform( - samples=samples, - name=name, - epsilon=header.epsilon, - limit_amplitude=header.amp_limited, - ) + common.data_from_binary(samples_raw, np.load) # read samples + value.read_value(file_obj, version, {}) # read name def _loads_obj(type_key, binary_data, version, vectors): @@ -77,26 +66,25 @@ def _loads_obj(type_key, binary_data, version, vectors): return value.loads_value(type_key, binary_data, version, vectors) -def _read_kernel(file_obj, version): - params = common.read_mapping( +def _read_kernel(file_obj, version) -> None: + common.read_mapping( file_obj=file_obj, deserializer=_loads_obj, version=version, vectors={}, ) - name = value.read_value(file_obj, version, {}) - return Kernel(name=name, **params) + value.read_value(file_obj, version, {}) # read name -def _read_discriminator(file_obj, version): - params = common.read_mapping( +def _read_discriminator(file_obj, version) -> None: + # read params + common.read_mapping( file_obj=file_obj, deserializer=_loads_obj, version=version, vectors={}, ) - name = value.read_value(file_obj, version, {}) - return Discriminator(name=name, **params) + value.read_value(file_obj, version, {}) # read name def _loads_symbolic_expr(expr_bytes, use_symengine=False): @@ -113,7 +101,7 @@ def _loads_symbolic_expr(expr_bytes, use_symengine=False): return sym.sympify(expr) -def _read_symbolic_pulse(file_obj, version): +def _read_symbolic_pulse(file_obj, version) -> None: make = formats.SYMBOLIC_PULSE._make pack = formats.SYMBOLIC_PULSE_PACK size = formats.SYMBOLIC_PULSE_SIZE @@ -125,10 +113,13 @@ def _read_symbolic_pulse(file_obj, version): ) ) pulse_type = file_obj.read(header.type_size).decode(common.ENCODE) - envelope = _loads_symbolic_expr(file_obj.read(header.envelope_size)) - constraints = _loads_symbolic_expr(file_obj.read(header.constraints_size)) - valid_amp_conditions = _loads_symbolic_expr(file_obj.read(header.valid_amp_conditions_size)) - parameters = common.read_mapping( + _loads_symbolic_expr(file_obj.read(header.envelope_size)) # read envelope + _loads_symbolic_expr(file_obj.read(header.constraints_size)) # read constraints + _loads_symbolic_expr( + file_obj.read(header.valid_amp_conditions_size) + ) # read valid amp conditions + # read parameters + common.read_mapping( file_obj, deserializer=value.loads_value, version=version, @@ -146,50 +137,16 @@ def _read_symbolic_pulse(file_obj, version): class_name = "SymbolicPulse" # Default class name, if not in the library if pulse_type in legacy_library_pulses: - parameters["angle"] = np.angle(parameters["amp"]) - parameters["amp"] = np.abs(parameters["amp"]) - _amp, _angle = sym.symbols("amp, angle") - envelope = envelope.subs(_amp, _amp * sym.exp(sym.I * _angle)) - - warnings.warn( - f"Library pulses with complex amp are no longer supported. " - f"{pulse_type} with complex amp was converted to (amp,angle) representation.", - UserWarning, - ) class_name = "ScalableSymbolicPulse" - duration = value.read_value(file_obj, version, {}) - name = value.read_value(file_obj, version, {}) - - if class_name == "SymbolicPulse": - return library.SymbolicPulse( - pulse_type=pulse_type, - duration=duration, - parameters=parameters, - name=name, - limit_amplitude=header.amp_limited, - envelope=envelope, - constraints=constraints, - valid_amp_conditions=valid_amp_conditions, - ) - elif class_name == "ScalableSymbolicPulse": - return library.ScalableSymbolicPulse( - pulse_type=pulse_type, - duration=duration, - amp=parameters["amp"], - angle=parameters["angle"], - parameters=parameters, - name=name, - limit_amplitude=header.amp_limited, - envelope=envelope, - constraints=constraints, - valid_amp_conditions=valid_amp_conditions, - ) - else: + value.read_value(file_obj, version, {}) # read duration + value.read_value(file_obj, version, {}) # read name + + if class_name not in {"SymbolicPulse", "ScalableSymbolicPulse"}: raise NotImplementedError(f"Unknown class '{class_name}'") -def _read_symbolic_pulse_v6(file_obj, version, use_symengine): +def _read_symbolic_pulse_v6(file_obj, version, use_symengine) -> None: make = formats.SYMBOLIC_PULSE_V2._make pack = formats.SYMBOLIC_PULSE_PACK_V2 size = formats.SYMBOLIC_PULSE_SIZE_V2 @@ -201,79 +158,36 @@ def _read_symbolic_pulse_v6(file_obj, version, use_symengine): ) ) class_name = file_obj.read(header.class_name_size).decode(common.ENCODE) - pulse_type = file_obj.read(header.type_size).decode(common.ENCODE) - envelope = _loads_symbolic_expr(file_obj.read(header.envelope_size), use_symengine) - constraints = _loads_symbolic_expr(file_obj.read(header.constraints_size), use_symengine) - valid_amp_conditions = _loads_symbolic_expr( + file_obj.read(header.type_size).decode(common.ENCODE) # read pulse type + _loads_symbolic_expr(file_obj.read(header.envelope_size), use_symengine) # read envelope + _loads_symbolic_expr(file_obj.read(header.constraints_size), use_symengine) # read constraints + _loads_symbolic_expr( file_obj.read(header.valid_amp_conditions_size), use_symengine - ) - parameters = common.read_mapping( + ) # read valid_amp_conditions + # read parameters + common.read_mapping( file_obj, deserializer=value.loads_value, version=version, vectors={}, ) - duration = value.read_value(file_obj, version, {}) - name = value.read_value(file_obj, version, {}) - - if class_name == "SymbolicPulse": - return library.SymbolicPulse( - pulse_type=pulse_type, - duration=duration, - parameters=parameters, - name=name, - limit_amplitude=header.amp_limited, - envelope=envelope, - constraints=constraints, - valid_amp_conditions=valid_amp_conditions, - ) - elif class_name == "ScalableSymbolicPulse": - # Between Qiskit 0.40 and 0.46, the (amp, angle) representation was present, - # but complex amp was still allowed. In Qiskit 1.0 and beyond complex amp - # is no longer supported and so the amp needs to be checked and converted. - # Once QPY version is bumped, a new reader function can be introduced without - # this check. - if isinstance(parameters["amp"], complex): - parameters["angle"] = np.angle(parameters["amp"]) - parameters["amp"] = np.abs(parameters["amp"]) - warnings.warn( - f"ScalableSymbolicPulse with complex amp are no longer supported. " - f"{pulse_type} with complex amp was converted to (amp,angle) representation.", - UserWarning, - ) + value.read_value(file_obj, version, {}) # read duration + value.read_value(file_obj, version, {}) # read name - return library.ScalableSymbolicPulse( - pulse_type=pulse_type, - duration=duration, - amp=parameters["amp"], - angle=parameters["angle"], - parameters=parameters, - name=name, - limit_amplitude=header.amp_limited, - envelope=envelope, - constraints=constraints, - valid_amp_conditions=valid_amp_conditions, - ) - else: + if class_name not in {"SymbolicPulse", "ScalableSymbolicPulse"}: raise NotImplementedError(f"Unknown class '{class_name}'") -def _read_alignment_context(file_obj, version): - type_key = common.read_type_key(file_obj) +def _read_alignment_context(file_obj, version) -> None: + common.read_type_key(file_obj) - context_params = common.read_sequence( + common.read_sequence( file_obj, deserializer=value.loads_value, version=version, vectors={}, ) - context_cls = type_keys.ScheduleAlignment.retrieve(type_key) - - instance = object.__new__(context_cls) - instance._context_params = tuple(context_params) - - return instance # pylint: disable=too-many-return-statements @@ -307,26 +221,23 @@ def _loads_operand(type_key, data_bytes, version, use_symengine): return value.loads_value(type_key, data_bytes, version, {}) -def _read_element(file_obj, version, metadata_deserializer, use_symengine): +def _read_element(file_obj, version, metadata_deserializer, use_symengine) -> None: type_key = common.read_type_key(file_obj) if type_key == type_keys.Program.SCHEDULE_BLOCK: return read_schedule_block(file_obj, version, metadata_deserializer, use_symengine) - operands = common.read_sequence( + # read operands + common.read_sequence( file_obj, deserializer=_loads_operand, version=version, use_symengine=use_symengine ) - name = value.read_value(file_obj, version, {}) + # read name + value.read_value(file_obj, version, {}) - instance = object.__new__(type_keys.ScheduleInstruction.retrieve(type_key)) - instance._operands = tuple(operands) - instance._name = name - instance._hash = None + return None - return instance - -def _loads_reference_item(type_key, data_bytes, metadata_deserializer, version): +def _loads_reference_item(type_key, data_bytes, metadata_deserializer, version) -> None: if type_key == type_keys.Value.NULL: return None if type_key == type_keys.Program.SCHEDULE_BLOCK: @@ -344,176 +255,8 @@ def _loads_reference_item(type_key, data_bytes, metadata_deserializer, version): ) -def _write_channel(file_obj, data, version): - type_key = type_keys.ScheduleChannel.assign(data) - common.write_type_key(file_obj, type_key) - value.write_value(file_obj, data.index, version=version) - - -def _write_waveform(file_obj, data, version): - samples_bytes = common.data_to_binary(data.samples, np.save) - - header = struct.pack( - formats.WAVEFORM_PACK, - data.epsilon, - len(samples_bytes), - data._limit_amplitude, - ) - file_obj.write(header) - file_obj.write(samples_bytes) - value.write_value(file_obj, data.name, version=version) - - -def _dumps_obj(obj, version): - """Wraps `value.dumps_value` to serialize dictionary and list objects - which are not supported by `value.dumps_value`. - """ - if isinstance(obj, dict): - with BytesIO() as container: - common.write_mapping( - file_obj=container, mapping=obj, serializer=_dumps_obj, version=version - ) - binary_data = container.getvalue() - return b"D", binary_data - elif isinstance(obj, list): - with BytesIO() as container: - common.write_sequence( - file_obj=container, sequence=obj, serializer=_dumps_obj, version=version - ) - binary_data = container.getvalue() - return b"l", binary_data - else: - return value.dumps_value(obj, version=version) - - -def _write_kernel(file_obj, data, version): - name = data.name - params = data.params - common.write_mapping(file_obj=file_obj, mapping=params, serializer=_dumps_obj, version=version) - value.write_value(file_obj, name, version=version) - - -def _write_discriminator(file_obj, data, version): - name = data.name - params = data.params - common.write_mapping(file_obj=file_obj, mapping=params, serializer=_dumps_obj, version=version) - value.write_value(file_obj, name, version=version) - - -def _dumps_symbolic_expr(expr, use_symengine): - if expr is None: - return b"" - if use_symengine: - expr_bytes = expr.__reduce__()[1][0] - else: - from sympy import srepr, sympify - - expr_bytes = srepr(sympify(expr)).encode(common.ENCODE) - return zlib.compress(expr_bytes) - - -def _write_symbolic_pulse(file_obj, data, use_symengine, version): - class_name_bytes = data.__class__.__name__.encode(common.ENCODE) - pulse_type_bytes = data.pulse_type.encode(common.ENCODE) - envelope_bytes = _dumps_symbolic_expr(data.envelope, use_symengine) - constraints_bytes = _dumps_symbolic_expr(data.constraints, use_symengine) - valid_amp_conditions_bytes = _dumps_symbolic_expr(data.valid_amp_conditions, use_symengine) - - header_bytes = struct.pack( - formats.SYMBOLIC_PULSE_PACK_V2, - len(class_name_bytes), - len(pulse_type_bytes), - len(envelope_bytes), - len(constraints_bytes), - len(valid_amp_conditions_bytes), - data._limit_amplitude, - ) - file_obj.write(header_bytes) - file_obj.write(class_name_bytes) - file_obj.write(pulse_type_bytes) - file_obj.write(envelope_bytes) - file_obj.write(constraints_bytes) - file_obj.write(valid_amp_conditions_bytes) - common.write_mapping( - file_obj, - mapping=data._params, - serializer=value.dumps_value, - version=version, - ) - value.write_value(file_obj, data.duration, version=version) - value.write_value(file_obj, data.name, version=version) - - -def _write_alignment_context(file_obj, context, version): - type_key = type_keys.ScheduleAlignment.assign(context) - common.write_type_key(file_obj, type_key) - common.write_sequence( - file_obj, sequence=context._context_params, serializer=value.dumps_value, version=version - ) - - -def _dumps_operand(operand, use_symengine, version): - if isinstance(operand, library.Waveform): - type_key = type_keys.ScheduleOperand.WAVEFORM - data_bytes = common.data_to_binary(operand, _write_waveform, version=version) - elif isinstance(operand, library.SymbolicPulse): - type_key = type_keys.ScheduleOperand.SYMBOLIC_PULSE - data_bytes = common.data_to_binary( - operand, _write_symbolic_pulse, use_symengine=use_symengine, version=version - ) - elif isinstance(operand, channels.Channel): - type_key = type_keys.ScheduleOperand.CHANNEL - data_bytes = common.data_to_binary(operand, _write_channel, version=version) - elif isinstance(operand, str): - type_key = type_keys.ScheduleOperand.OPERAND_STR - data_bytes = operand.encode(common.ENCODE) - elif isinstance(operand, Kernel): - type_key = type_keys.ScheduleOperand.KERNEL - data_bytes = common.data_to_binary(operand, _write_kernel, version=version) - elif isinstance(operand, Discriminator): - type_key = type_keys.ScheduleOperand.DISCRIMINATOR - data_bytes = common.data_to_binary(operand, _write_discriminator, version=version) - else: - type_key, data_bytes = value.dumps_value(operand, version=version) - - return type_key, data_bytes - - -def _write_element(file_obj, element, metadata_serializer, use_symengine, version): - if isinstance(element, ScheduleBlock): - common.write_type_key(file_obj, type_keys.Program.SCHEDULE_BLOCK) - write_schedule_block(file_obj, element, metadata_serializer, use_symengine, version=version) - else: - type_key = type_keys.ScheduleInstruction.assign(element) - common.write_type_key(file_obj, type_key) - common.write_sequence( - file_obj, - sequence=element.operands, - serializer=_dumps_operand, - use_symengine=use_symengine, - version=version, - ) - value.write_value(file_obj, element.name, version=version) - - -def _dumps_reference_item(schedule, metadata_serializer, version): - if schedule is None: - type_key = type_keys.Value.NULL - data_bytes = b"" - else: - type_key = type_keys.Program.SCHEDULE_BLOCK - data_bytes = common.data_to_binary( - obj=schedule, - serializer=write_schedule_block, - metadata_serializer=metadata_serializer, - version=version, - ) - return type_key, data_bytes - - -@ignore_pulse_deprecation_warnings def read_schedule_block(file_obj, version, metadata_deserializer=None, use_symengine=False): - """Read a single ScheduleBlock from the file like object. + """Consume a single ScheduleBlock from the file like object. Args: file_obj (File): A file like object that contains the QPY binary data. @@ -530,7 +273,9 @@ def read_schedule_block(file_obj, version, metadata_deserializer=None, use_symen platforms. Please check that your target platform is supported by the symengine library before setting this option, as it will be required by qpy to deserialize the payload. Returns: - ScheduleBlock: The schedule block object from the file. + QuantumCircuit: Returns a dummy QuantumCircuit object, containing just name and metadata. + This function exists just to allow reading legacy payloads containing pulse information + without breaking the entire load flow. Raises: TypeError: If any of the instructions is invalid data format. @@ -545,91 +290,19 @@ def read_schedule_block(file_obj, version, metadata_deserializer=None, use_symen file_obj.read(formats.SCHEDULE_BLOCK_HEADER_SIZE), ) ) - name = file_obj.read(data.name_size).decode(common.ENCODE) + file_obj.read(data.name_size).decode(common.ENCODE) # read name metadata_raw = file_obj.read(data.metadata_size) - metadata = json.loads(metadata_raw, cls=metadata_deserializer) - context = _read_alignment_context(file_obj, version) + json.loads(metadata_raw, cls=metadata_deserializer) # read metadata + _read_alignment_context(file_obj, version) - block = ScheduleBlock( - name=name, - metadata=metadata, - alignment_context=context, - ) for _ in range(data.num_elements): - block_elm = _read_element(file_obj, version, metadata_deserializer, use_symengine) - block.append(block_elm, inplace=True) + _read_element(file_obj, version, metadata_deserializer, use_symengine) # Load references if version >= 7: - flat_key_refdict = common.read_mapping( + common.read_mapping( file_obj=file_obj, deserializer=_loads_reference_item, version=version, metadata_deserializer=metadata_deserializer, ) - ref_dict = {} - for key_str, schedule in flat_key_refdict.items(): - if schedule is not None: - composite_key = tuple(key_str.split(instructions.Reference.key_delimiter)) - ref_dict[composite_key] = schedule - if ref_dict: - block.assign_references(ref_dict, inplace=True) - - return block - - -def write_schedule_block( - file_obj, block, metadata_serializer=None, use_symengine=False, version=common.QPY_VERSION -): - """Write a single ScheduleBlock object in the file like object. - - Args: - file_obj (File): The file like object to write the circuit data in. - block (ScheduleBlock): A schedule block data to write. - metadata_serializer (JSONEncoder): An optional JSONEncoder class that - will be passed the :attr:`.ScheduleBlock.metadata` dictionary for - ``block`` and will be used as the ``cls`` kwarg - on the ``json.dump()`` call to JSON serialize that dictionary. - use_symengine (bool): If True, symbolic objects will be serialized using symengine's - native mechanism. This is a faster serialization alternative, but not supported in all - platforms. Please check that your target platform is supported by the symengine library - before setting this option, as it will be required by qpy to deserialize the payload. - version (int): The QPY format version to use for serializing this circuit block - Raises: - TypeError: If any of the instructions is invalid data format. - """ - metadata = json.dumps(block.metadata, separators=(",", ":"), cls=metadata_serializer).encode( - common.ENCODE - ) - block_name = block.name.encode(common.ENCODE) - - # Write schedule block header - header_raw = formats.SCHEDULE_BLOCK_HEADER( - name_size=len(block_name), - metadata_size=len(metadata), - num_elements=len(block), - ) - header = struct.pack(formats.SCHEDULE_BLOCK_HEADER_PACK, *header_raw) - file_obj.write(header) - file_obj.write(block_name) - file_obj.write(metadata) - - _write_alignment_context(file_obj, block.alignment_context, version=version) - for block_elm in block._blocks: - # Do not call block.blocks. This implicitly assigns references to instruction. - # This breaks original reference structure. - _write_element(file_obj, block_elm, metadata_serializer, use_symengine, version=version) - - # Write references - flat_key_refdict = {} - for ref_keys, schedule in block._reference_manager.items(): - # Do not call block.reference. This returns the reference of most outer program by design. - key_str = instructions.Reference.key_delimiter.join(ref_keys) - flat_key_refdict[key_str] = schedule - common.write_mapping( - file_obj=file_obj, - mapping=flat_key_refdict, - serializer=_dumps_reference_item, - metadata_serializer=metadata_serializer, - version=version, - ) diff --git a/qiskit/qpy/interface.py b/qiskit/qpy/interface.py index cab90eb9407f..f518d15ae32c 100644 --- a/qiskit/qpy/interface.py +++ b/qiskit/qpy/interface.py @@ -22,16 +22,14 @@ import re from qiskit.circuit import QuantumCircuit -from qiskit.pulse import ScheduleBlock from qiskit.exceptions import QiskitError from qiskit.qpy import formats, common, binary_io, type_keys -from qiskit.qpy.exceptions import QPYLoadingDeprecatedFeatureWarning, QpyError +from qiskit.qpy.exceptions import QpyError from qiskit.version import __version__ -from qiskit.utils.deprecate_pulse import deprecate_pulse_arg # pylint: disable=invalid-name -QPY_SUPPORTED_TYPES = Union[QuantumCircuit, ScheduleBlock] +QPY_SUPPORTED_TYPES = QuantumCircuit # This version pattern is taken from the pypa packaging project: # https://github.com/pypa/packaging/blob/21.3/packaging/version.py#L223-L254 @@ -74,11 +72,6 @@ VERSION_PATTERN_REGEX = re.compile(VERSION_PATTERN, re.VERBOSE | re.IGNORECASE) -@deprecate_pulse_arg( - "programs", - deprecation_description="Passing `ScheduleBlock` to `programs`", - predicate=lambda p: isinstance(p, ScheduleBlock), -) def dump( programs: Union[List[QPY_SUPPORTED_TYPES], QPY_SUPPORTED_TYPES], file_obj: BinaryIO, @@ -127,9 +120,7 @@ def dump( Args: programs: QPY supported object(s) to store in the specified file like object. - QPY supports :class:`.QuantumCircuit` and :class:`.ScheduleBlock`. - Different data types must be separately serialized. - Support for :class:`.ScheduleBlock` is deprecated since Qiskit 1.3.0. + QPY supports :class:`.QuantumCircuit`. file_obj: The file like object to write the QPY data too metadata_serializer: An optional JSONEncoder class that will be passed the ``.metadata`` attribute for each program in ``programs`` and will be @@ -156,7 +147,7 @@ def dump( .. note:: - If serializing a :class:`.QuantumCircuit` or :class:`.ScheduleBlock` that contain + If serializing a :class:`.QuantumCircuit` that contains :class:`.ParameterExpression` objects with ``version`` set low with the intent to load the payload using a historical release of Qiskit, it is safest to set the ``use_symengine`` flag to ``False``. Versions of Qiskit prior to 1.2.4 cannot load @@ -166,32 +157,19 @@ def dump( Raises: - QpyError: When multiple data format is mixed in the output. TypeError: When invalid data type is input. - ValueError: When an unsupported version number is passed in for the ``version`` argument + ValueError: When an unsupported version number is passed in for the ``version`` argument. """ if not isinstance(programs, Iterable): programs = [programs] - program_types = set() + # dump accepts only QuantumCircuit typed objects for program in programs: - program_types.add(type(program)) - - if len(program_types) > 1: - raise QpyError( - "Input programs contain multiple data types. " - "Different data type must be serialized separately." - ) - program_type = next(iter(program_types)) + if not issubclass(type(program), QuantumCircuit): + raise TypeError(f"'{type(program)}' is not a supported data type.") - if issubclass(program_type, QuantumCircuit): - type_key = type_keys.Program.CIRCUIT - writer = binary_io.write_circuit - elif program_type is ScheduleBlock: - type_key = type_keys.Program.SCHEDULE_BLOCK - writer = binary_io.write_schedule_block - else: - raise TypeError(f"'{program_type}' is not supported data type.") + type_key = type_keys.Program.CIRCUIT + writer = binary_io.write_circuit if version is None: version = common.QPY_VERSION @@ -218,10 +196,7 @@ def dump( file_obj.write(header) common.write_type_key(file_obj, type_key) - pulse_gates = False for program in programs: - if type_key == type_keys.Program.CIRCUIT and program._calibrations_prop: - pulse_gates = True writer( file_obj, program, @@ -230,13 +205,6 @@ def dump( version=version, ) - if pulse_gates: - warnings.warn( - category=DeprecationWarning, - message="Pulse gates serialization is deprecated as of Qiskit 1.3. " - "It will be removed in Qiskit 2.0.", - ) - def load( file_obj: BinaryIO, @@ -245,8 +213,7 @@ def load( """Load a QPY binary file This function is used to load a serialized QPY Qiskit program file and create - :class:`~qiskit.circuit.QuantumCircuit` objects or - :class:`~qiskit.pulse.schedule.ScheduleBlock` objects from its contents. + :class:`~qiskit.circuit.QuantumCircuit` objects from its contents. For example: .. code-block:: python @@ -267,12 +234,11 @@ def load( circuits = qpy.load(fd) which will read the contents of the qpy and return a list of - :class:`~qiskit.circuit.QuantumCircuit` objects or - :class:`~qiskit.pulse.schedule.ScheduleBlock` objects from the file. + :class:`~qiskit.circuit.QuantumCircuit` objects from the file. Args: file_obj: A file like object that contains the QPY binary - data for a circuit or pulse schedule. + data for a circuit. metadata_deserializer: An optional JSONDecoder class that will be used for the ``cls`` kwarg on the internal ``json.load`` call used to deserialize the JSON payload used for @@ -286,8 +252,9 @@ def load( A list is always returned, even if there is only 1 program in the QPY data. Raises: - QiskitError: if ``file_obj`` is not a valid QPY file - TypeError: When invalid data type is loaded. + QiskitError: if ``file_obj`` is not a valid QPY file. + QpyError: if known but unsupported data type is loaded. + TypeError: if invalid data type is loaded. """ # identify file header version @@ -350,15 +317,10 @@ def load( if type_key == type_keys.Program.CIRCUIT: loader = binary_io.read_circuit elif type_key == type_keys.Program.SCHEDULE_BLOCK: - loader = binary_io.read_schedule_block - warnings.warn( - category=QPYLoadingDeprecatedFeatureWarning, - message="Pulse gates deserialization is deprecated as of Qiskit 1.3 and " - "will be removed in Qiskit 2.0. This is part of the deprecation plan for " - "the entire Qiskit Pulse package. Once Pulse is removed, `ScheduleBlock` " - "sections will be ignored when loading QPY files with pulse data.", + raise QpyError( + "Payloads of type `ScheduleBlock` cannot be loaded as of Qiskit 2.0. " + "Use an earlier version of Qiskit if you want to load `ScheduleBlock` payloads." ) - else: raise TypeError(f"Invalid payload format data kind '{type_key}'.") diff --git a/qiskit/qpy/type_keys.py b/qiskit/qpy/type_keys.py index 60262440d033..b4dc6c4cd465 100644 --- a/qiskit/qpy/type_keys.py +++ b/qiskit/qpy/type_keys.py @@ -37,36 +37,6 @@ from qiskit.circuit.parameter import Parameter from qiskit.circuit.parameterexpression import ParameterExpression from qiskit.circuit.parametervector import ParameterVectorElement -from qiskit.pulse.channels import ( - Channel, - DriveChannel, - MeasureChannel, - ControlChannel, - AcquireChannel, - MemorySlot, - RegisterSlot, -) -from qiskit.pulse.configuration import Discriminator, Kernel -from qiskit.pulse.instructions import ( - Acquire, - Play, - Delay, - SetFrequency, - ShiftFrequency, - SetPhase, - ShiftPhase, - RelativeBarrier, - TimeBlockade, - Reference, -) -from qiskit.pulse.library import Waveform, SymbolicPulse -from qiskit.pulse.schedule import ScheduleBlock -from qiskit.pulse.transforms.alignments import ( - AlignLeft, - AlignRight, - AlignSequential, - AlignEquispaced, -) from qiskit.qpy import exceptions @@ -168,7 +138,7 @@ class Condition(IntEnum): class Container(TypeKeyBase): - """Typle key enum for container-like object.""" + """Type key enum for container-like object.""" RANGE = b"r" TUPLE = b"t" @@ -220,125 +190,13 @@ def retrieve(cls, type_key): raise NotImplementedError -class ScheduleAlignment(TypeKeyBase): - """Type key enum for schedule block alignment context object.""" - - LEFT = b"l" - RIGHT = b"r" - SEQUENTIAL = b"s" - EQUISPACED = b"e" - - # AlignFunc is not serializable due to the callable in context parameter - - @classmethod - def assign(cls, obj): - if isinstance(obj, AlignLeft): - return cls.LEFT - if isinstance(obj, AlignRight): - return cls.RIGHT - if isinstance(obj, AlignSequential): - return cls.SEQUENTIAL - if isinstance(obj, AlignEquispaced): - return cls.EQUISPACED - - raise exceptions.QpyError( - f"Object type '{type(obj)}' is not supported in {cls.__name__} namespace." - ) - - @classmethod - def retrieve(cls, type_key): - if type_key == cls.LEFT: - return AlignLeft - if type_key == cls.RIGHT: - return AlignRight - if type_key == cls.SEQUENTIAL: - return AlignSequential - if type_key == cls.EQUISPACED: - return AlignEquispaced - - raise exceptions.QpyError( - f"A class corresponding to type key '{type_key}' is not found in {cls.__name__} namespace." - ) - - -class ScheduleInstruction(TypeKeyBase): - """Type key enum for schedule instruction object.""" - - ACQUIRE = b"a" - PLAY = b"p" - DELAY = b"d" - SET_FREQUENCY = b"f" - SHIFT_FREQUENCY = b"g" - SET_PHASE = b"q" - SHIFT_PHASE = b"r" - BARRIER = b"b" - TIME_BLOCKADE = b"t" - REFERENCE = b"y" - - # 's' is reserved by ScheduleBlock, i.e. block can be nested as an element. - # Call instruction is not supported by QPY. - # This instruction has been excluded from ScheduleBlock instructions with - # qiskit-terra/#8005 and new instruction Reference will be added instead. - # Call is only applied to Schedule which is not supported by QPY. - # Also snapshot is not suppored because of its limited usecase. - - @classmethod - def assign(cls, obj): - if isinstance(obj, Acquire): - return cls.ACQUIRE - if isinstance(obj, Play): - return cls.PLAY - if isinstance(obj, Delay): - return cls.DELAY - if isinstance(obj, SetFrequency): - return cls.SET_FREQUENCY - if isinstance(obj, ShiftFrequency): - return cls.SHIFT_FREQUENCY - if isinstance(obj, SetPhase): - return cls.SET_PHASE - if isinstance(obj, ShiftPhase): - return cls.SHIFT_PHASE - if isinstance(obj, RelativeBarrier): - return cls.BARRIER - if isinstance(obj, TimeBlockade): - return cls.TIME_BLOCKADE - if isinstance(obj, Reference): - return cls.REFERENCE - - raise exceptions.QpyError( - f"Object type '{type(obj)}' is not supported in {cls.__name__} namespace." - ) - - @classmethod - def retrieve(cls, type_key): - if type_key == cls.ACQUIRE: - return Acquire - if type_key == cls.PLAY: - return Play - if type_key == cls.DELAY: - return Delay - if type_key == cls.SET_FREQUENCY: - return SetFrequency - if type_key == cls.SHIFT_FREQUENCY: - return ShiftFrequency - if type_key == cls.SET_PHASE: - return SetPhase - if type_key == cls.SHIFT_PHASE: - return ShiftPhase - if type_key == cls.BARRIER: - return RelativeBarrier - if type_key == cls.TIME_BLOCKADE: - return TimeBlockade - if type_key == cls.REFERENCE: - return Reference - - raise exceptions.QpyError( - f"A class corresponding to type key '{type_key}' is not found in {cls.__name__} namespace." - ) - - class ScheduleOperand(TypeKeyBase): - """Type key enum for schedule instruction operand object.""" + """Type key enum for schedule instruction operand object. + + Note: This class is kept post pulse-removal to allow reading of + legacy payloads containing pulse gates without breaking the entire + load flow. + """ WAVEFORM = b"w" SYMBOLIC_PULSE = b"s" @@ -353,92 +211,27 @@ class ScheduleOperand(TypeKeyBase): OPERAND_STR = b"o" @classmethod - def assign(cls, obj): - if isinstance(obj, Waveform): - return cls.WAVEFORM - if isinstance(obj, SymbolicPulse): - return cls.SYMBOLIC_PULSE - if isinstance(obj, Channel): - return cls.CHANNEL - if isinstance(obj, str): - return cls.OPERAND_STR - if isinstance(obj, Kernel): - return cls.KERNEL - if isinstance(obj, Discriminator): - return cls.DISCRIMINATOR - - raise exceptions.QpyError( - f"Object type '{type(obj)}' is not supported in {cls.__name__} namespace." - ) - - @classmethod - def retrieve(cls, type_key): + def assign(cls, _): raise NotImplementedError - -class ScheduleChannel(TypeKeyBase): - """Type key enum for schedule channel object.""" - - DRIVE = b"d" - CONTROL = b"c" - MEASURE = b"m" - ACQURE = b"a" - MEM_SLOT = b"e" - REG_SLOT = b"r" - - # SnapShot channel is not defined because of its limited usecase. - @classmethod - def assign(cls, obj): - if isinstance(obj, DriveChannel): - return cls.DRIVE - if isinstance(obj, ControlChannel): - return cls.CONTROL - if isinstance(obj, MeasureChannel): - return cls.MEASURE - if isinstance(obj, AcquireChannel): - return cls.ACQURE - if isinstance(obj, MemorySlot): - return cls.MEM_SLOT - if isinstance(obj, RegisterSlot): - return cls.REG_SLOT - - raise exceptions.QpyError( - f"Object type '{type(obj)}' is not supported in {cls.__name__} namespace." - ) - - @classmethod - def retrieve(cls, type_key): - if type_key == cls.DRIVE: - return DriveChannel - if type_key == cls.CONTROL: - return ControlChannel - if type_key == cls.MEASURE: - return MeasureChannel - if type_key == cls.ACQURE: - return AcquireChannel - if type_key == cls.MEM_SLOT: - return MemorySlot - if type_key == cls.REG_SLOT: - return RegisterSlot - - raise exceptions.QpyError( - f"A class corresponding to type key '{type_key}' is not found in {cls.__name__} namespace." - ) + def retrieve(cls, _): + raise NotImplementedError class Program(TypeKeyBase): - """Typle key enum for program that QPY supports.""" + """Type key enum for program that QPY supports.""" CIRCUIT = b"q" + # This is left for backward compatibility, for identifying payloads of type `ScheduleBlock` + # and raising accordingly. `ScheduleBlock` support has been removed in Qiskit 2.0 as part + # of the pulse package removal in that version. SCHEDULE_BLOCK = b"s" @classmethod def assign(cls, obj): if isinstance(obj, QuantumCircuit): return cls.CIRCUIT - if isinstance(obj, ScheduleBlock): - return cls.SCHEDULE_BLOCK raise exceptions.QpyError( f"Object type '{type(obj)}' is not supported in {cls.__name__} namespace." diff --git a/releasenotes/notes/remove-pulse-qpy-07a96673c8f10e38.yaml b/releasenotes/notes/remove-pulse-qpy-07a96673c8f10e38.yaml new file mode 100644 index 000000000000..7b1ff7a150a6 --- /dev/null +++ b/releasenotes/notes/remove-pulse-qpy-07a96673c8f10e38.yaml @@ -0,0 +1,8 @@ +--- +upgrade_qpy: + - | + With the removal of Pulse in Qiskit 2.0, support for serializing ``ScheduleBlock`` programs + via the :func:`qiskit.qpy.dump` function has been removed. Users can still load payloads + containing pulse gates using the :func:`qiskit.qpy.load` function, however those will be + treated as opaque custom instructions. Loading ``ScheduleBlock`` payloads is not supported + anymore and will result with a :class:`.QpyError` exception. diff --git a/test/python/circuit/test_circuit_load_from_qpy.py b/test/python/circuit/test_circuit_load_from_qpy.py index 63d256649518..3fff7ddf3e02 100644 --- a/test/python/circuit/test_circuit_load_from_qpy.py +++ b/test/python/circuit/test_circuit_load_from_qpy.py @@ -21,7 +21,7 @@ import ddt import numpy as np -from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister, pulse +from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister from qiskit.circuit import CASE_DEFAULT, IfElseOp, WhileLoopOp, SwitchCaseOp from qiskit.circuit.classical import expr, types from qiskit.circuit.classicalregister import Clbit @@ -318,44 +318,6 @@ def test_bound_parameter(self): self.assertEqual(qc, new_circ) self.assertDeprecatedBitProperties(qc, new_circ) - def test_bound_calibration_parameter(self): - """Test a circuit with a bound calibration parameter is correctly serialized. - - In particular, this test ensures that parameters on a circuit - instruction are consistent with the circuit's calibrations dictionary - after serialization. - """ - amp = Parameter("amp") - - with self.assertWarns(DeprecationWarning): - with pulse.builder.build() as sched: - pulse.builder.play(pulse.Constant(100, amp), pulse.DriveChannel(0)) - - gate = Gate("custom", 1, [amp]) - - qc = QuantumCircuit(1) - qc.append(gate, (0,)) - with self.assertWarns(DeprecationWarning): - qc.add_calibration(gate, (0,), sched) - qc.assign_parameters({amp: 1 / 3}, inplace=True) - - qpy_file = io.BytesIO() - with self.assertWarns(DeprecationWarning): - # qpy.dump warns for deprecations of pulse gate serialization - dump(qc, qpy_file) - qpy_file.seek(0) - new_circ = load(qpy_file)[0] - self.assertEqual(qc, new_circ) - instruction = new_circ.data[0] - cal_key = ( - tuple(new_circ.find_bit(q).index for q in instruction.qubits), - tuple(instruction.operation.params), - ) - # Make sure that looking for a calibration based on the instruction's - # parameters succeeds - with self.assertWarns(DeprecationWarning): - self.assertIn(cal_key, new_circ.calibrations[gate.name]) - def test_parameter_expression(self): """Test a circuit with a parameter expression.""" theta = Parameter("theta") diff --git a/test/python/qpy/test_block_load_from_qpy.py b/test/python/qpy/test_block_load_from_qpy.py deleted file mode 100644 index 6085618e8362..000000000000 --- a/test/python/qpy/test_block_load_from_qpy.py +++ /dev/null @@ -1,521 +0,0 @@ -# This code is part of Qiskit. -# -# (C) Copyright IBM 2022. -# -# This code is licensed under the Apache License, Version 2.0. You may -# obtain a copy of this license in the LICENSE.txt file in the root directory -# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. -# -# Any modifications or derivative works of this code must retain this -# copyright notice, and modified files need to carry a notice indicating -# that they have been altered from the originals. - -"""Test cases for the schedule block qpy loading and saving.""" - -import io -import unittest -import warnings -from ddt import ddt, data, unpack -import numpy as np -import symengine as sym - -from qiskit.pulse import builder, Schedule -from qiskit.pulse.library import ( - SymbolicPulse, - Gaussian, - GaussianSquare, - Drag, - Constant, - Waveform, -) -from qiskit.pulse.channels import ( - DriveChannel, - ControlChannel, - MeasureChannel, - AcquireChannel, - MemorySlot, - RegisterSlot, -) -from qiskit.pulse.instructions import Play, TimeBlockade -from qiskit.circuit import Parameter, QuantumCircuit, Gate -from qiskit.qpy import dump, load -from qiskit.qpy.exceptions import QPYLoadingDeprecatedFeatureWarning -from qiskit.utils import optionals as _optional -from qiskit.pulse.configuration import Kernel, Discriminator -from test import QiskitTestCase # pylint: disable=wrong-import-order - - -class QpyScheduleTestCase(QiskitTestCase): - """QPY schedule testing platform.""" - - def assert_roundtrip_equal(self, block, use_symengine=False): - """QPY roundtrip equal test.""" - qpy_file = io.BytesIO() - with self.assertWarns(DeprecationWarning): - dump(block, qpy_file, use_symengine=use_symengine) - qpy_file.seek(0) - new_block = load(qpy_file)[0] - - self.assertEqual(block, new_block) - - -@ddt -class TestLoadFromQPY(QpyScheduleTestCase): - """Test loading and saving schedule block to qpy file.""" - - @data( - (Gaussian, DriveChannel, 160, 0.1, 40), - (GaussianSquare, DriveChannel, 800, 0.1, 64, 544), - (Drag, DriveChannel, 160, 0.1, 40, 0.5), - (Constant, DriveChannel, 800, 0.1), - (Constant, ControlChannel, 800, 0.1), - (Constant, MeasureChannel, 800, 0.1), - ) - @unpack - def test_library_pulse_play(self, envelope, channel, *params): - """Test playing standard pulses.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.play( - envelope(*params), - channel(0), - ) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_playing_custom_symbolic_pulse(self): - """Test playing a custom user pulse.""" - # pylint: disable=invalid-name - t, amp, freq = sym.symbols("t, amp, freq") - sym_envelope = 2 * amp * (freq * t - sym.floor(1 / 2 + freq * t)) - - with self.assertWarns(DeprecationWarning): - my_pulse = SymbolicPulse( - pulse_type="Sawtooth", - duration=100, - parameters={"amp": 0.1, "freq": 0.05}, - envelope=sym_envelope, - name="pulse1", - ) - with builder.build() as test_sched: - builder.play(my_pulse, DriveChannel(0)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_symbolic_amplitude_limit(self): - """Test applying amplitude limit to symbolic pulse.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.play( - Gaussian(160, 20, 40, limit_amplitude=False), - DriveChannel(0), - ) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_waveform_amplitude_limit(self): - """Test applying amplitude limit to waveform.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.play( - Waveform([1, 2, 3, 4, 5], limit_amplitude=False), - DriveChannel(0), - ) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_playing_waveform(self): - """Test playing waveform.""" - # pylint: disable=invalid-name - t = np.linspace(0, 1, 100) - waveform = 0.1 * np.sin(2 * np.pi * t) - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.play(waveform, DriveChannel(0)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_phases(self): - """Test phase.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.shift_phase(0.1, DriveChannel(0)) - builder.set_phase(0.4, DriveChannel(1)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_frequencies(self): - """Test frequency.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.shift_frequency(10e6, DriveChannel(0)) - builder.set_frequency(5e9, DriveChannel(1)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_delay(self): - """Test delay.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.delay(100, DriveChannel(0)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_barrier(self): - """Test barrier.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.barrier(DriveChannel(0), DriveChannel(1), ControlChannel(2)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_time_blockade(self): - """Test time blockade.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.append_instruction(TimeBlockade(10, DriveChannel(0))) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_measure(self): - """Test measurement.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.acquire(100, AcquireChannel(0), MemorySlot(0)) - builder.acquire(100, AcquireChannel(1), RegisterSlot(1)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - @data( - (0, Parameter("dur"), 0.1, 40), - (Parameter("ch1"), 160, 0.1, 40), - (Parameter("ch1"), Parameter("dur"), Parameter("amp"), Parameter("sigma")), - (0, 160, Parameter("amp") * np.exp(1j * Parameter("phase")), 40), - ) - @unpack - def test_parameterized(self, channel, *params): - """Test playing parameterized pulse.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.play(Gaussian(*params), DriveChannel(channel)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_nested_blocks(self): - """Test nested blocks with different alignment contexts.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - with builder.align_equispaced(duration=1200): - with builder.align_left(): - builder.delay(100, DriveChannel(0)) - builder.delay(200, DriveChannel(1)) - with builder.align_right(): - builder.delay(100, DriveChannel(0)) - builder.delay(200, DriveChannel(1)) - with builder.align_sequential(): - builder.delay(100, DriveChannel(0)) - builder.delay(200, DriveChannel(1)) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_called_schedule(self): - """Test referenced pulse Schedule object. - - Referenced object is naively converted into ScheduleBlock with TimeBlockade instructions. - Thus referenced Schedule is still QPY compatible. - """ - with self.assertWarns(DeprecationWarning): - refsched = Schedule() - refsched.insert(20, Play(Constant(100, 0.1), DriveChannel(0))) - refsched.insert(50, Play(Constant(100, 0.1), DriveChannel(1))) - - with builder.build() as test_sched: - builder.call(refsched, name="test_ref") - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_unassigned_reference(self): - """Test schedule with unassigned reference.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.reference("custom1", "q0") - builder.reference("custom1", "q1") - - with warnings.catch_warnings(): - warnings.simplefilter(action="ignore", category=DeprecationWarning) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_partly_assigned_reference(self): - """Test schedule with partly assigned reference.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.reference("custom1", "q0") - builder.reference("custom1", "q1") - - with builder.build() as sub_q0: - builder.delay(Parameter("duration"), DriveChannel(0)) - - test_sched.assign_references( - {("custom1", "q0"): sub_q0}, - inplace=True, - ) - - with warnings.catch_warnings(): - warnings.simplefilter(action="ignore", category=DeprecationWarning) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_nested_assigned_reference(self): - """Test schedule with assigned reference for nested schedule.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - with builder.align_left(): - builder.reference("custom1", "q0") - builder.reference("custom1", "q1") - - with builder.build() as sub_q0: - builder.delay(Parameter("duration"), DriveChannel(0)) - - with builder.build() as sub_q1: - builder.delay(Parameter("duration"), DriveChannel(1)) - - test_sched.assign_references( - {("custom1", "q0"): sub_q0, ("custom1", "q1"): sub_q1}, - inplace=True, - ) - - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_bell_schedule(self): - """Test complex schedule to create a Bell state.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - with builder.align_sequential(): - # H - builder.shift_phase(-1.57, DriveChannel(0)) - builder.play(Drag(160, 0.05, 40, 1.3), DriveChannel(0)) - builder.shift_phase(-1.57, DriveChannel(0)) - # ECR - with builder.align_left(): - builder.play(GaussianSquare(800, 0.05, 64, 544), DriveChannel(1)) - builder.play(GaussianSquare(800, 0.22, 64, 544, 2), ControlChannel(0)) - builder.play(Drag(160, 0.1, 40, 1.5), DriveChannel(0)) - with builder.align_left(): - builder.play(GaussianSquare(800, -0.05, 64, 544), DriveChannel(1)) - builder.play(GaussianSquare(800, -0.22, 64, 544, 2), ControlChannel(0)) - builder.play(Drag(160, 0.1, 40, 1.5), DriveChannel(0)) - # Measure - with builder.align_left(): - builder.play(GaussianSquare(8000, 0.2, 64, 7744), MeasureChannel(0)) - builder.acquire(8000, AcquireChannel(0), MemorySlot(0)) - - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - @unittest.skipUnless(_optional.HAS_SYMENGINE, "Symengine required for this test") - def test_bell_schedule_use_symengine(self): - """Test complex schedule to create a Bell state.""" - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - with builder.align_sequential(): - # H - builder.shift_phase(-1.57, DriveChannel(0)) - builder.play(Drag(160, 0.05, 40, 1.3), DriveChannel(0)) - builder.shift_phase(-1.57, DriveChannel(0)) - # ECR - with builder.align_left(): - builder.play(GaussianSquare(800, 0.05, 64, 544), DriveChannel(1)) - builder.play(GaussianSquare(800, 0.22, 64, 544, 2), ControlChannel(0)) - builder.play(Drag(160, 0.1, 40, 1.5), DriveChannel(0)) - with builder.align_left(): - builder.play(GaussianSquare(800, -0.05, 64, 544), DriveChannel(1)) - builder.play(GaussianSquare(800, -0.22, 64, 544, 2), ControlChannel(0)) - builder.play(Drag(160, 0.1, 40, 1.5), DriveChannel(0)) - # Measure - with builder.align_left(): - builder.play(GaussianSquare(8000, 0.2, 64, 7744), MeasureChannel(0)) - builder.acquire(8000, AcquireChannel(0), MemorySlot(0)) - - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched, True) - - def test_with_acquire_instruction_with_kernel(self): - """Test a schedblk with acquire instruction with kernel.""" - kernel = Kernel( - name="my_kernel", kernel={"real": np.ones(10), "imag": np.zeros(10)}, bias=[0, 0] - ) - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.acquire(100, AcquireChannel(0), MemorySlot(0), kernel=kernel) - - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - def test_with_acquire_instruction_with_discriminator(self): - """Test a schedblk with acquire instruction with a discriminator.""" - discriminator = Discriminator( - name="my_discriminator", discriminator_type="linear", params=[1, 0] - ) - with self.assertWarns(DeprecationWarning): - with builder.build() as test_sched: - builder.acquire(100, AcquireChannel(0), MemorySlot(0), discriminator=discriminator) - - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - self.assert_roundtrip_equal(test_sched) - - -class TestPulseGate(QpyScheduleTestCase): - """Test loading and saving pulse gate attached circuit to qpy file.""" - - def test_1q_gate(self): - """Test for single qubit pulse gate.""" - mygate = Gate("mygate", 1, []) - - with self.assertWarns(DeprecationWarning): - with builder.build() as caldef: - builder.play(Constant(100, 0.1), DriveChannel(0)) - - qc = QuantumCircuit(2) - qc.append(mygate, [0]) - with self.assertWarns(DeprecationWarning): - qc.add_calibration(mygate, (0,), caldef) - - self.assert_roundtrip_equal(qc) - - def test_2q_gate(self): - """Test for two qubit pulse gate.""" - mygate = Gate("mygate", 2, []) - - with self.assertWarns(DeprecationWarning): - with builder.build() as caldef: - builder.play(Constant(100, 0.1), ControlChannel(0)) - - qc = QuantumCircuit(2) - qc.append(mygate, [0, 1]) - with self.assertWarns(DeprecationWarning): - qc.add_calibration(mygate, (0, 1), caldef) - - self.assert_roundtrip_equal(qc) - - def test_parameterized_gate(self): - """Test for parameterized pulse gate.""" - amp = Parameter("amp") - angle = Parameter("angle") - mygate = Gate("mygate", 2, [amp, angle]) - - with self.assertWarns(DeprecationWarning): - with builder.build() as caldef: - builder.play(Constant(100, amp * np.exp(1j * angle)), ControlChannel(0)) - - qc = QuantumCircuit(2) - qc.append(mygate, [0, 1]) - with self.assertWarns(DeprecationWarning): - qc.add_calibration(mygate, (0, 1), caldef) - - self.assert_roundtrip_equal(qc) - - def test_override(self): - """Test for overriding standard gate with pulse gate.""" - amp = Parameter("amp") - - with self.assertWarns(DeprecationWarning): - with builder.build() as caldef: - builder.play(Constant(100, amp), ControlChannel(0)) - - qc = QuantumCircuit(2) - qc.rx(amp, 0) - with self.assertWarns(DeprecationWarning): - qc.add_calibration("rx", (0,), caldef, [amp]) - - self.assert_roundtrip_equal(qc) - - def test_multiple_calibrations(self): - """Test for circuit with multiple pulse gates.""" - amp1 = Parameter("amp1") - amp2 = Parameter("amp2") - mygate = Gate("mygate", 1, [amp2]) - - with self.assertWarns(DeprecationWarning): - with builder.build() as caldef1: - builder.play(Constant(100, amp1), DriveChannel(0)) - - with builder.build() as caldef2: - builder.play(Constant(100, amp2), DriveChannel(1)) - - qc = QuantumCircuit(2) - qc.rx(amp1, 0) - qc.append(mygate, [1]) - with self.assertWarns(DeprecationWarning): - qc.add_calibration("rx", (0,), caldef1, [amp1]) - qc.add_calibration(mygate, (1,), caldef2) - - self.assert_roundtrip_equal(qc) - - def test_with_acquire_instruction_with_kernel(self): - """Test a pulse gate with acquire instruction with kernel.""" - kernel = Kernel( - name="my_kernel", kernel={"real": np.zeros(10), "imag": np.zeros(10)}, bias=[0, 0] - ) - - with self.assertWarns(DeprecationWarning): - with builder.build() as sched: - builder.acquire(10, AcquireChannel(0), MemorySlot(0), kernel=kernel) - - qc = QuantumCircuit(1, 1) - qc.measure(0, 0) - with self.assertWarns(DeprecationWarning): - qc.add_calibration("measure", (0,), sched) - - self.assert_roundtrip_equal(qc) - - def test_with_acquire_instruction_with_discriminator(self): - """Test a pulse gate with acquire instruction with discriminator.""" - discriminator = Discriminator("my_discriminator") - - with self.assertWarns(DeprecationWarning): - with builder.build() as sched: - builder.acquire(10, AcquireChannel(0), MemorySlot(0), discriminator=discriminator) - - qc = QuantumCircuit(1, 1) - qc.measure(0, 0) - with self.assertWarns(DeprecationWarning): - qc.add_calibration("measure", (0,), sched) - - self.assert_roundtrip_equal(qc) - - -class TestSymengineLoadFromQPY(QiskitTestCase): - """Test use of symengine in qpy set of methods.""" - - def setUp(self): - super().setUp() - - # pylint: disable=invalid-name - t, amp, freq = sym.symbols("t, amp, freq") - sym_envelope = 2 * amp * (freq * t - sym.floor(1 / 2 + freq * t)) - - with self.assertWarns(DeprecationWarning): - my_pulse = SymbolicPulse( - pulse_type="Sawtooth", - duration=100, - parameters={"amp": 0.1, "freq": 0.05}, - envelope=sym_envelope, - name="pulse1", - ) - with builder.build() as test_sched: - builder.play(my_pulse, DriveChannel(0)) - - self.test_sched = test_sched - - @unittest.skipIf(not _optional.HAS_SYMENGINE, "Install symengine to run this test.") - def test_symengine_full_path(self): - """Test use_symengine option for circuit with parameter expressions.""" - qpy_file = io.BytesIO() - with self.assertWarns(DeprecationWarning): - dump(self.test_sched, qpy_file, use_symengine=True) - qpy_file.seek(0) - with self.assertWarns(QPYLoadingDeprecatedFeatureWarning): - new_sched = load(qpy_file)[0] - self.assertEqual(self.test_sched, new_sched) diff --git a/test/python/qpy/test_circuit_load_from_qpy.py b/test/python/qpy/test_circuit_load_from_qpy.py index c95e54857759..a2d83d755f85 100644 --- a/test/python/qpy/test_circuit_load_from_qpy.py +++ b/test/python/qpy/test_circuit_load_from_qpy.py @@ -10,7 +10,7 @@ # copyright notice, and modified files need to carry a notice indicating # that they have been altered from the originals. -"""Test cases for the schedule block qpy loading and saving.""" +"""Test cases for circuit qpy loading and saving.""" import io import struct @@ -30,7 +30,7 @@ class QpyCircuitTestCase(QiskitTestCase): - """QPY schedule testing platform.""" + """QPY circuit testing platform.""" def assert_roundtrip_equal(self, circuit, version=None, use_symengine=None): """QPY roundtrip equal test.""" diff --git a/test/qpy_compat/test_qpy.py b/test/qpy_compat/test_qpy.py index 0c064587b9e2..eeca4bb041da 100755 --- a/test/qpy_compat/test_qpy.py +++ b/test/qpy_compat/test_qpy.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 + # This code is part of Qiskit. # # (C) Copyright IBM 2021. @@ -440,6 +441,7 @@ def generate_control_flow_switch_circuits(): def generate_schedule_blocks(current_version): """Standard QPY testcase for schedule blocks.""" + # pylint: disable=no-name-in-module from qiskit.pulse import builder, channels, library # Parameterized schedule test is avoided. @@ -501,6 +503,7 @@ def generate_schedule_blocks(current_version): def generate_referenced_schedule(): """Test for QPY serialization of unassigned reference schedules.""" + # pylint: disable=no-name-in-module from qiskit.pulse import builder, channels, library schedule_blocks = [] @@ -526,6 +529,7 @@ def generate_referenced_schedule(): def generate_calibrated_circuits(): """Test for QPY serialization with calibrations.""" + # pylint: disable=no-name-in-module from qiskit.pulse import builder, Constant, DriveChannel circuits = [] @@ -596,6 +600,7 @@ def generate_open_controlled_gates(): def generate_acquire_instruction_with_kernel_and_discriminator(): """Test QPY serialization with Acquire instruction with kernel and discriminator.""" + # pylint: disable=no-name-in-module from qiskit.pulse import builder, AcquireChannel, MemorySlot, Discriminator, Kernel schedule_blocks = [] @@ -828,8 +833,13 @@ def generate_v12_expr(): return [index, shift] -def generate_circuits(version_parts, current_version): - """Generate reference circuits.""" +def generate_circuits(version_parts, current_version, load_context=False): + """Generate reference circuits. + + If load_context is True, avoid generating Pulse-based reference + circuits. For those circuits, load_qpy only checks that the cached + circuits can be loaded without erroring.""" + output_circuits = { "full.qpy": [generate_full_circuit()], "unitary.qpy": [generate_unitary_gate_circuit()], @@ -858,20 +868,27 @@ def generate_circuits(version_parts, current_version): ] if version_parts >= (0, 19, 2): output_circuits["control_flow.qpy"] = generate_control_flow_circuits() - if version_parts >= (0, 21, 0): - output_circuits["schedule_blocks.qpy"] = generate_schedule_blocks(current_version) - output_circuits["pulse_gates.qpy"] = generate_calibrated_circuits() + if version_parts >= (0, 21, 0) and version_parts < (2, 0): + output_circuits["schedule_blocks.qpy"] = ( + None if load_context else generate_schedule_blocks(current_version) + ) + output_circuits["pulse_gates.qpy"] = ( + None if load_context else generate_calibrated_circuits() + ) + if version_parts >= (0, 24, 0) and version_parts < (2, 0): + output_circuits["referenced_schedule_blocks.qpy"] = ( + None if load_context else generate_referenced_schedule() + ) if version_parts >= (0, 24, 0): - output_circuits["referenced_schedule_blocks.qpy"] = generate_referenced_schedule() output_circuits["control_flow_switch.qpy"] = generate_control_flow_switch_circuits() if version_parts >= (0, 24, 1): output_circuits["open_controlled_gates.qpy"] = generate_open_controlled_gates() output_circuits["controlled_gates.qpy"] = generate_controlled_gates() if version_parts >= (0, 24, 2): output_circuits["layout.qpy"] = generate_layout_circuits() - if version_parts >= (0, 25, 0): + if version_parts >= (0, 25, 0) and version_parts < (2, 0): output_circuits["acquire_inst_with_kernel_and_disc.qpy"] = ( - generate_acquire_instruction_with_kernel_and_discriminator() + None if load_context else generate_acquire_instruction_with_kernel_and_discriminator() ) output_circuits["control_flow_expr.qpy"] = generate_control_flow_expr() if version_parts >= (0, 45, 2): @@ -960,7 +977,19 @@ def generate_qpy(qpy_files): def load_qpy(qpy_files, version_parts): """Load qpy circuits from files and compare to reference circuits.""" + pulse_files = { + "schedule_blocks.qpy": (0, 21, 0), + "pulse_gates.qpy": (0, 21, 0), + "referenced_schedule_blocks.qpy": (0, 24, 0), + "acquire_inst_with_kernel_and_disc.qpy": (0, 25, 0), + } for path, circuits in qpy_files.items(): + if path in pulse_files.keys(): + # Qiskit Pulse was removed in version 2.0. Loading ScheduleBlock payloads + # raises an exception and loading pulse gates results with undefined instructions + # so not loading and comparing these payloads. + # See https://github.com/Qiskit/qiskit/pull/13814 + continue print(f"Loading qpy file: {path}") with open(path, "rb") as fd: qpy_circuits = load(fd) @@ -983,6 +1012,34 @@ def load_qpy(qpy_files, version_parts): circuit, qpy_circuits[i], i, version_parts, bind=bind, equivalent=equivalent ) + from qiskit.qpy.exceptions import QpyError + + while pulse_files: + path, version = pulse_files.popitem() + + if version_parts < version or version_parts >= (2, 0): + continue + + if path == "pulse_gates.qpy": + try: + with open(path, "rb") as fd: + load(fd) + except: + msg = f"Loading circuit with pulse gates should not raise" + sys.stderr.write(msg) + sys.exit(1) + else: + try: + # A ScheduleBlock payload, should raise QpyError + with open(path, "rb") as fd: + load(fd) + except QpyError: + continue + + msg = f"Loading payload {path} didn't raise QpyError" + sys.stderr.write(msg) + sys.exit(1) + def _main(): parser = argparse.ArgumentParser(description="Test QPY backwards compatibility") @@ -1011,10 +1068,11 @@ def _main(): version_match = re.search(VERSION_PATTERN, args.version, re.VERBOSE | re.IGNORECASE) version_parts = tuple(int(x) for x in version_match.group("release").split(".")) - qpy_files = generate_circuits(version_parts, current_version) if args.command == "generate": + qpy_files = generate_circuits(version_parts, current_version) generate_qpy(qpy_files) else: + qpy_files = generate_circuits(version_parts, current_version, load_context=True) load_qpy(qpy_files, version_parts)