# This code is part of Qiskit.
#
# (C) Copyright IBM 2017, 2019.
#
# 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.
"""Assemble function for converting a list of circuits into a qobj."""
import hashlib
from collections import defaultdict
from typing import Any, Dict, List, Tuple, Union
from qiskit import qobj, pulse
from qiskit.assembler.run_config import RunConfig
from qiskit.exceptions import QiskitError
from qiskit.pulse import instructions, transforms, library, schedule, channels
from qiskit.qobj import utils as qobj_utils, converters
from qiskit.qobj.converters.pulse_instruction import ParametricPulseShapes
[문서]def assemble_schedules(
schedules: List[
Union[
schedule.ScheduleBlock,
schedule.ScheduleComponent,
Tuple[int, schedule.ScheduleComponent],
]
],
qobj_id: int,
qobj_header: qobj.QobjHeader,
run_config: RunConfig,
) -> qobj.PulseQobj:
"""Assembles a list of schedules into a qobj that can be run on the backend.
Args:
schedules: Schedules to assemble.
qobj_id: Identifier for the generated qobj.
qobj_header: Header to pass to the results.
run_config: Configuration of the runtime environment.
Returns:
The Qobj to be run on the backends.
Raises:
QiskitError: when frequency settings are not supplied.
Examples:
.. code-block:: python
from qiskit import pulse
from qiskit.assembler import assemble_schedules
from qiskit.assembler.run_config import RunConfig
# Construct a Qobj header for the output Qobj
header = {"backend_name": "FakeOpenPulse2Q", "backend_version": "0.0.0"}
# Build a configuration object for the output Qobj
config = RunConfig(shots=1024,
memory=False,
meas_level=1,
meas_return='avg',
memory_slot_size=100,
parametric_pulses=[],
init_qubits=True,
qubit_lo_freq=[4900000000.0, 5000000000.0],
meas_lo_freq=[6500000000.0, 6600000000.0],
schedule_los=[])
# Build a Pulse schedule to assemble into a Qobj
schedule = pulse.Schedule()
schedule += pulse.Play(pulse.Waveform([0.1] * 16, name="test0"),
pulse.DriveChannel(0),
name="test1")
schedule += pulse.Play(pulse.Waveform([0.1] * 16, name="test1"),
pulse.DriveChannel(0),
name="test2")
schedule += pulse.Play(pulse.Waveform([0.5] * 16, name="test0"),
pulse.DriveChannel(0),
name="test1")
# Assemble a Qobj from the schedule.
pulseQobj = assemble_schedules(schedules=[schedule],
qobj_id="custom-id",
qobj_header=header,
run_config=config)
"""
if not hasattr(run_config, "qubit_lo_freq"):
raise QiskitError("qubit_lo_freq must be supplied.")
if not hasattr(run_config, "meas_lo_freq"):
raise QiskitError("meas_lo_freq must be supplied.")
lo_converter = converters.LoConfigConverter(
qobj.PulseQobjExperimentConfig, **run_config.to_dict()
)
experiments, experiment_config = _assemble_experiments(schedules, lo_converter, run_config)
qobj_config = _assemble_config(lo_converter, experiment_config, run_config)
return qobj.PulseQobj(
experiments=experiments, qobj_id=qobj_id, header=qobj_header, config=qobj_config
)
def _assemble_experiments(
schedules: List[Union[schedule.ScheduleComponent, Tuple[int, schedule.ScheduleComponent]]],
lo_converter: converters.LoConfigConverter,
run_config: RunConfig,
) -> Tuple[List[qobj.PulseQobjExperiment], Dict[str, Any]]:
"""Assembles a list of schedules into PulseQobjExperiments, and returns related metadata that
will be assembled into the Qobj configuration.
Args:
schedules: Schedules to assemble.
lo_converter: The configured frequency converter and validator.
run_config: Configuration of the runtime environment.
Returns:
The list of assembled experiments, and the dictionary of related experiment config.
Raises:
QiskitError: when frequency settings are not compatible with the experiments.
"""
freq_configs = [lo_converter(lo_dict) for lo_dict in getattr(run_config, "schedule_los", [])]
if len(schedules) > 1 and len(freq_configs) not in [0, 1, len(schedules)]:
raise QiskitError(
"Invalid 'schedule_los' setting specified. If specified, it should be "
"either have a single entry to apply the same LOs for each schedule or "
"have length equal to the number of schedules."
)
instruction_converter = getattr(
run_config, "instruction_converter", converters.InstructionToQobjConverter
)
instruction_converter = instruction_converter(qobj.PulseQobjInstruction, **run_config.to_dict())
formatted_schedules = [transforms.target_qobj_transform(sched) for sched in schedules]
compressed_schedules = transforms.compress_pulses(formatted_schedules)
user_pulselib = {}
experiments = []
for idx, sched in enumerate(compressed_schedules):
qobj_instructions, max_memory_slot = _assemble_instructions(
sched, instruction_converter, run_config, user_pulselib
)
metadata = sched.metadata
if metadata is None:
metadata = {}
# TODO: add other experimental header items (see circuit assembler)
qobj_experiment_header = qobj.QobjExperimentHeader(
memory_slots=max_memory_slot + 1, # Memory slots are 0 indexed
name=sched.name or "Experiment-%d" % idx,
metadata=metadata,
)
experiment = qobj.PulseQobjExperiment(
header=qobj_experiment_header, instructions=qobj_instructions
)
if freq_configs:
# This handles the cases where one frequency setting applies to all experiments and
# where each experiment has a different frequency
freq_idx = idx if len(freq_configs) != 1 else 0
experiment.config = freq_configs[freq_idx]
experiments.append(experiment)
# Frequency sweep
if freq_configs and len(experiments) == 1:
experiment = experiments[0]
experiments = []
for freq_config in freq_configs:
experiments.append(
qobj.PulseQobjExperiment(
header=experiment.header,
instructions=experiment.instructions,
config=freq_config,
)
)
# Top level Qobj configuration
experiment_config = {
"pulse_library": [
qobj.PulseLibraryItem(name=name, samples=samples)
for name, samples in user_pulselib.items()
],
"memory_slots": max(exp.header.memory_slots for exp in experiments),
}
return experiments, experiment_config
def _assemble_instructions(
sched: Union[pulse.Schedule, pulse.ScheduleBlock],
instruction_converter: converters.InstructionToQobjConverter,
run_config: RunConfig,
user_pulselib: Dict[str, List[complex]],
) -> Tuple[List[qobj.PulseQobjInstruction], int]:
"""Assembles the instructions in a schedule into a list of PulseQobjInstructions and returns
related metadata that will be assembled into the Qobj configuration. Lookup table for
pulses defined in all experiments are registered in ``user_pulselib``. This object should be
mutable python dictionary so that items are properly updated after each instruction assemble.
The dictionary is not returned to avoid redundancy.
Args:
sched: Schedule to assemble.
instruction_converter: A converter instance which can convert PulseInstructions to
PulseQobjInstructions.
run_config: Configuration of the runtime environment.
user_pulselib: User pulse library from previous schedule.
Returns:
A list of converted instructions, the user pulse library dictionary (from pulse name to
pulse samples), and the maximum number of readout memory slots used by this Schedule.
"""
sched = transforms.target_qobj_transform(sched)
max_memory_slot = 0
qobj_instructions = []
acquire_instruction_map = defaultdict(list)
for time, instruction in sched.instructions:
if isinstance(instruction, instructions.Play):
if isinstance(instruction.pulse, (library.ParametricPulse, library.SymbolicPulse)):
is_backend_supported = True
try:
pulse_shape = ParametricPulseShapes.from_instance(instruction.pulse).name
if pulse_shape not in run_config.parametric_pulses:
is_backend_supported = False
except ValueError:
# Custom pulse class, or bare SymbolicPulse object.
is_backend_supported = False
if not is_backend_supported:
instruction = instructions.Play(
instruction.pulse.get_waveform(), instruction.channel, name=instruction.name
)
if isinstance(instruction.pulse, library.Waveform):
name = hashlib.sha256(instruction.pulse.samples).hexdigest()
instruction = instructions.Play(
library.Waveform(name=name, samples=instruction.pulse.samples),
channel=instruction.channel,
name=name,
)
user_pulselib[name] = instruction.pulse.samples
# ignore explicit delay instrs on acq channels as they are invalid on IBMQ backends;
# timing of other instrs will still be shifted appropriately
if isinstance(instruction, instructions.Delay) and isinstance(
instruction.channel, channels.AcquireChannel
):
continue
if isinstance(instruction, instructions.Acquire):
if instruction.mem_slot:
max_memory_slot = max(max_memory_slot, instruction.mem_slot.index)
# Acquires have a single AcquireChannel per inst, but we have to bundle them
# together into the Qobj as one instruction with many channels
acquire_instruction_map[(time, instruction.duration)].append(instruction)
continue
qobj_instructions.append(instruction_converter(time, instruction))
if acquire_instruction_map:
if hasattr(run_config, "meas_map"):
_validate_meas_map(acquire_instruction_map, run_config.meas_map)
for (time, _), instruction_bundle in acquire_instruction_map.items():
qobj_instructions.append(
instruction_converter(time, instruction_bundle),
)
return qobj_instructions, max_memory_slot
def _validate_meas_map(
instruction_map: Dict[Tuple[int, instructions.Acquire], List[instructions.Acquire]],
meas_map: List[List[int]],
) -> None:
"""Validate all qubits tied in ``meas_map`` are to be acquired.
Args:
instruction_map: A dictionary grouping Acquire instructions according to their start time
and duration.
meas_map: List of groups of qubits that must be acquired together.
Raises:
QiskitError: If the instructions do not satisfy the measurement map.
"""
sorted_inst_map = sorted(instruction_map.items(), key=lambda item: item[0])
meas_map_sets = [set(m) for m in meas_map]
# error if there is time overlap between qubits in the same meas_map
for idx, inst in enumerate(sorted_inst_map[:-1]):
inst_end_time = inst[0][0] + inst[0][1]
next_inst = sorted_inst_map[idx + 1]
next_inst_time = next_inst[0][0]
if next_inst_time < inst_end_time:
inst_qubits = {inst.channel.index for inst in inst[1]}
next_inst_qubits = {inst.channel.index for inst in next_inst[1]}
for meas_set in meas_map_sets:
common_instr_qubits = inst_qubits.intersection(meas_set)
common_next = next_inst_qubits.intersection(meas_set)
if common_instr_qubits and common_next:
raise QiskitError(
"Qubits {} and {} are in the same measurement grouping: {}. "
"They must either be acquired at the same time, or disjointly"
". Instead, they were acquired at times: {}-{} and "
"{}-{}".format(
common_instr_qubits,
common_next,
meas_map,
inst[0][0],
inst_end_time,
next_inst_time,
next_inst_time + next_inst[0][1],
)
)
def _assemble_config(
lo_converter: converters.LoConfigConverter,
experiment_config: Dict[str, Any],
run_config: RunConfig,
) -> qobj.PulseQobjConfig:
"""Assembles the QobjConfiguration from experimental config and runtime config.
Args:
lo_converter: The configured frequency converter and validator.
experiment_config: Schedules to assemble.
run_config: Configuration of the runtime environment.
Returns:
The assembled PulseQobjConfig.
"""
qobj_config = run_config.to_dict()
qobj_config.update(experiment_config)
# Run config not needed in qobj config
qobj_config.pop("meas_map", None)
qobj_config.pop("qubit_lo_range", None)
qobj_config.pop("meas_lo_range", None)
# convert enums to serialized values
meas_return = qobj_config.get("meas_return", "avg")
if isinstance(meas_return, qobj_utils.MeasReturnType):
qobj_config["meas_return"] = meas_return.value
meas_level = qobj_config.get("meas_level", 2)
if isinstance(meas_level, qobj_utils.MeasLevel):
qobj_config["meas_level"] = meas_level.value
# convert LO frequencies to GHz
qobj_config["qubit_lo_freq"] = [freq / 1e9 for freq in qobj_config["qubit_lo_freq"]]
qobj_config["meas_lo_freq"] = [freq / 1e9 for freq in qobj_config["meas_lo_freq"]]
# override defaults if single entry for ``schedule_los``
schedule_los = qobj_config.pop("schedule_los", [])
if len(schedule_los) == 1:
lo_dict = schedule_los[0]
q_los = lo_converter.get_qubit_los(lo_dict)
# Hz -> GHz
if q_los:
qobj_config["qubit_lo_freq"] = [freq / 1e9 for freq in q_los]
m_los = lo_converter.get_meas_los(lo_dict)
if m_los:
qobj_config["meas_lo_freq"] = [freq / 1e9 for freq in m_los]
return qobj.PulseQobjConfig(**qobj_config)