mirror of https://github.com/Qiskit/qiskit.git
Pulse command definition (#2263)
* Add pulse backend configuration. * Add QASM and Pulse backend configuration objects. * Added QASMBackendConfiguration and PulseBackendConfiguration with factory method from_dict. * add tests for qasm/pulse backend configuration. * linting. * Parameter update * Cleanup provider models and rename. * linting * Linting * Make QASM -> Qasm * remove invalid name * Update open_pulse field to required. * Remove from_dict hackery. * Update tests. * linting. * linting. * linting and required variable. * readd comment. * Add setting of u_channel_lo frequency. * Redo qubit signature, all control channels present. * linting * Remove lo_freq from pulse channels. * linting. * Rework the way lo configuration works. Removed from channels and instead responsibility is placed on assembler. * Small change to trigger rebuild. * Fix rebasing. * Add buffering of commands. * Add buffer tests. * linting. * Small bug in device specification fall back. * Add buffer to acquire channel. * Added command definition data structure. * Added casting of integer input qubits to tuple for cmd_def. * Added cmd_def object tests. Not running as proper pulse schedule api is not yet running. * Update imports and making pulse library. * Added cmd_def and pulse_library to defaults in pulse mock backend * Remove duplicate of PulseLibraryItem. * remove redefinition of pulse qobj schema. * linting * update schema for parameterizable pulses. * Add mutable mapping * Remove mutablemapping and dunder methods. * cmd_def pulse_library defaults methods. * Add cmd_def accesor method. * Update schema for vals and phase of type string. * Added parameterized schedule and typing to cmd_def. * Update typing of CmdDef. Add cmd_qubits method. * Make ConversionMethodBinder generic. * Rename PulseQobjConverter to InstructionToQobjConverter * Add conversion logic for qobj to schedules. * Convert without built pulse_library. * QobjToInstruction tests and passing. * Small bug fix. * Fix cmd_def tests. * linting. * linting. * add linting. * Add safe expression parsing. * Add Parameterized Schedule logic. * Add parameterized schedules with tests. * Fix cmd_def test. * Working cmd def. * Added parameterized cmd_def test * Update tests. * Tests passing. * change cmd_types to cmds) * Update cmd_def errors. * Update error strings. * Fix cmd_qubits. * Linting. * update regex and docs. * Add testing of sanitization * Update sanitization testing. * linting. * Update imports. * Undo standard imports. * Fix bug in pulse * Escape regex
This commit is contained in:
Thomas Alexander
Kevin Krsulich
parent
9b93d17877
commit
52b9d37f17
|
|
@ -19,8 +19,8 @@ from qiskit.exceptions import QiskitError
|
|||
from qiskit.pulse.commands import PulseInstruction
|
||||
from qiskit.qobj import (PulseQobj, QobjExperimentHeader,
|
||||
PulseQobjInstruction, PulseQobjExperimentConfig,
|
||||
PulseQobjExperiment, PulseQobjConfig, QobjPulseLibrary)
|
||||
from qiskit.qobj.converters import PulseQobjConverter, LoConfigConverter
|
||||
PulseQobjExperiment, PulseQobjConfig, PulseLibraryItem)
|
||||
from qiskit.qobj.converters import InstructionToQobjConverter, LoConfigConverter
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -40,7 +40,7 @@ def assemble_schedules(schedules, qobj_id=None, qobj_header=None, run_config=Non
|
|||
if hasattr(run_config, 'instruction_converter'):
|
||||
instruction_converter = run_config.instruction_converter
|
||||
else:
|
||||
instruction_converter = PulseQobjConverter
|
||||
instruction_converter = InstructionToQobjConverter
|
||||
|
||||
qobj_config = run_config.to_dict()
|
||||
qubit_lo_range = qobj_config.pop('qubit_lo_range')
|
||||
|
|
@ -75,7 +75,7 @@ def assemble_schedules(schedules, qobj_id=None, qobj_header=None, run_config=Non
|
|||
})
|
||||
|
||||
# setup pulse_library
|
||||
qobj_config['pulse_library'] = [QobjPulseLibrary(name=pulse.name, samples=pulse.samples)
|
||||
qobj_config['pulse_library'] = [PulseLibraryItem(name=pulse.name, samples=pulse.samples)
|
||||
for pulse in user_pulselib]
|
||||
|
||||
# create qob experiment field
|
||||
|
|
|
|||
|
|
@ -15,8 +15,8 @@
|
|||
"""Qiskit schema-conformant objects used by the backends and providers."""
|
||||
|
||||
from .backendconfiguration import (BackendConfiguration, PulseBackendConfiguration,
|
||||
QasmBackendConfiguration, UchannelLO)
|
||||
QasmBackendConfiguration, UchannelLO, GateConfig)
|
||||
from .backendproperties import BackendProperties
|
||||
from .backendstatus import BackendStatus
|
||||
from .jobstatus import JobStatus
|
||||
from .pulsedefaults import PulseDefaults
|
||||
from .pulsedefaults import PulseDefaults, Command
|
||||
|
|
|
|||
|
|
@ -11,17 +11,9 @@ from marshmallow.validate import Length, Range
|
|||
|
||||
from qiskit.validation import BaseModel, BaseSchema, bind_schema
|
||||
from qiskit.validation.base import ObjSchema
|
||||
from qiskit.validation.fields import (Complex, Integer, List, Nested, Number,
|
||||
String)
|
||||
|
||||
|
||||
class PulseLibraryItemSchema(BaseSchema):
|
||||
"""Schema for PulseLibraryItem."""
|
||||
|
||||
# Required properties.
|
||||
name = String(required=True)
|
||||
samples = List(Complex(), required=True,
|
||||
validate=Length(min=1))
|
||||
from qiskit.validation.fields import (Integer, List, Nested, Number, String)
|
||||
from qiskit.qobj import PulseLibraryItemSchema, PulseQobjInstructionSchema
|
||||
from qiskit.pulse import CmdDef
|
||||
|
||||
|
||||
class MeasurementKernelSchema(BaseSchema):
|
||||
|
|
@ -49,7 +41,7 @@ class CommandSchema(BaseSchema):
|
|||
# Optional properties.
|
||||
qubits = List(Integer(validate=Range(min=0)),
|
||||
validate=Length(min=1))
|
||||
sequence = Nested(ObjSchema, many=True)
|
||||
sequence = Nested(PulseQobjInstructionSchema, many=True)
|
||||
|
||||
|
||||
class PulseDefaultsSchema(BaseSchema):
|
||||
|
|
@ -67,24 +59,6 @@ class PulseDefaultsSchema(BaseSchema):
|
|||
discriminator = Nested(DiscriminatorSchema)
|
||||
|
||||
|
||||
@bind_schema(PulseLibraryItemSchema)
|
||||
class PulseLibraryItem(BaseModel):
|
||||
"""Model for PulseLibraryItem.
|
||||
|
||||
Please note that this class only describes the required fields. For the
|
||||
full description of the model, please check ``PulseLibraryItemSchema``.
|
||||
|
||||
Attributes:
|
||||
name (str): Pulse name.
|
||||
samples (list[complex]): Pulse samples.
|
||||
"""
|
||||
def __init__(self, name, samples, **kwargs):
|
||||
self.name = name
|
||||
self.samples = samples
|
||||
|
||||
super().__init__(**kwargs)
|
||||
|
||||
|
||||
@bind_schema(MeasurementKernelSchema)
|
||||
class MeasurementKernel(BaseModel):
|
||||
"""Model for MeasurementKernel.
|
||||
|
|
@ -146,3 +120,11 @@ class PulseDefaults(BaseModel):
|
|||
self.cmd_def = cmd_def
|
||||
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def build_cmd_def(self):
|
||||
"""Construct the `CmdDef` object for the backend.
|
||||
|
||||
Returns:
|
||||
CmdDef: `CmdDef` instance generated from defaults
|
||||
"""
|
||||
return CmdDef.from_defaults(self.cmd_def, self.pulse_library)
|
||||
|
|
|
|||
|
|
@ -19,3 +19,5 @@ from .commands import (Acquire, FrameChange, PersistentValue, SamplePulse, Snaps
|
|||
Kernel, Discriminator, functional_pulse)
|
||||
from .configuration import LoConfig, LoRange
|
||||
from .schedule import Schedule
|
||||
from .cmd_def import CmdDef
|
||||
from .exceptions import PulseError
|
||||
|
|
|
|||
|
|
@ -0,0 +1,192 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2019, IBM.
|
||||
#
|
||||
# This source code is licensed under the Apache License, Version 2.0 found in
|
||||
# the LICENSE.txt file in the root directory of this source tree.
|
||||
|
||||
"""
|
||||
Command definition module. Relates circuit gates to pulse commands.
|
||||
"""
|
||||
from typing import List, Tuple, Iterable, Union, Dict
|
||||
|
||||
from qiskit.qobj import PulseQobjInstruction
|
||||
from qiskit.qobj.converters import QobjToInstructionConverter
|
||||
|
||||
from .commands import SamplePulse
|
||||
|
||||
from .exceptions import PulseError
|
||||
from .schedule import Schedule, ParameterizedSchedule
|
||||
|
||||
# pylint: disable=missing-return-doc
|
||||
|
||||
|
||||
def _to_qubit_tuple(qubit_tuple: Union[int, Iterable[int]]) -> Tuple[int]:
|
||||
"""Convert argument to tuple.
|
||||
Args:
|
||||
qubit_tuple: Qubits to enforce as tuple.
|
||||
|
||||
Raises:
|
||||
PulseError: If qubits are not integers
|
||||
"""
|
||||
try:
|
||||
qubit_tuple = tuple(qubit_tuple)
|
||||
except TypeError:
|
||||
qubit_tuple = (qubit_tuple,)
|
||||
|
||||
if not all(isinstance(i, int) for i in qubit_tuple):
|
||||
raise PulseError("All qubits must be integers.")
|
||||
|
||||
return qubit_tuple
|
||||
|
||||
|
||||
class CmdDef:
|
||||
"""Command definition class. Relates `Gate`s to `Schedule`s."""
|
||||
|
||||
def __init__(self, schedules: Dict = None):
|
||||
"""Create command definition from backend.
|
||||
|
||||
Args:
|
||||
schedules: Keys are tuples of (cmd_name, *qubits) and values are
|
||||
`Schedule` or `ParameterizedSchedule`
|
||||
"""
|
||||
self._cmd_dict = {}
|
||||
|
||||
if schedules:
|
||||
for key, schedule in schedules.items():
|
||||
self.add(key[0], key[1:], schedule)
|
||||
|
||||
@classmethod
|
||||
def from_defaults(cls, flat_cmd_def: List[PulseQobjInstruction],
|
||||
pulse_library: Dict[str, SamplePulse]) -> 'CmdDef':
|
||||
"""Create command definition from backend defaults output.
|
||||
Args:
|
||||
flat_cmd_def: Command definition list returned by backend
|
||||
pulse_library: Dictionary of `SamplePulse`s
|
||||
"""
|
||||
converter = QobjToInstructionConverter(pulse_library, buffer=0)
|
||||
cmd_def = cls()
|
||||
|
||||
for cmd in flat_cmd_def:
|
||||
qubits = cmd.qubits
|
||||
name = cmd.name
|
||||
instructions = []
|
||||
for instr in cmd.sequence:
|
||||
instructions.append(converter(instr))
|
||||
|
||||
cmd_def.add(name, qubits, ParameterizedSchedule(*instructions, name=name))
|
||||
|
||||
return cmd_def
|
||||
|
||||
def add(self, cmd_name: str, qubits: Union[int, Iterable[int]],
|
||||
schedule: Union[ParameterizedSchedule, Schedule]):
|
||||
"""Add a command to the `CommandDefinition`
|
||||
|
||||
Args:
|
||||
cmd_name: Name of the command
|
||||
qubits: Qubits command applies to
|
||||
schedule: Schedule to be added
|
||||
"""
|
||||
qubits = _to_qubit_tuple(qubits)
|
||||
cmd_dict = self._cmd_dict.setdefault(cmd_name, {})
|
||||
if isinstance(schedule, Schedule):
|
||||
schedule = ParameterizedSchedule(schedule, name=schedule.name)
|
||||
cmd_dict[qubits] = schedule
|
||||
|
||||
def has(self, cmd_name: str, qubits: Union[int, Iterable[int]]) -> bool:
|
||||
"""Has command of name with qubits.
|
||||
|
||||
Args:
|
||||
cmd_name: Name of the command
|
||||
qubits: Ordered list of qubits command applies to
|
||||
"""
|
||||
qubits = _to_qubit_tuple(qubits)
|
||||
if cmd_name in self._cmd_dict:
|
||||
|
||||
if qubits in self._cmd_dict[cmd_name]:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def get(self, cmd_name: str, qubits: Union[int, Iterable[int]],
|
||||
**params: Dict[str, Union[float, complex]]) -> Schedule:
|
||||
"""Get command from command definition.
|
||||
Args:
|
||||
cmd_name: Name of the command
|
||||
qubits: Ordered list of qubits command applies to
|
||||
**params: Command parameters to be used to generate schedule
|
||||
|
||||
Raises:
|
||||
PulseError: If command for qubits is not available
|
||||
"""
|
||||
qubits = _to_qubit_tuple(qubits)
|
||||
if self.has(cmd_name, qubits):
|
||||
schedule = self._cmd_dict[cmd_name][qubits]
|
||||
|
||||
if isinstance(schedule, ParameterizedSchedule):
|
||||
return schedule.bind_parameters(**params)
|
||||
|
||||
return schedule.flatten()
|
||||
|
||||
else:
|
||||
raise PulseError('Command {0} for qubits {1} is not present '
|
||||
'in CmdDef'.format(cmd_name, qubits))
|
||||
|
||||
def get_parameters(self, cmd_name: str, qubits: Union[int, Iterable[int]]) -> Tuple[str]:
|
||||
"""Get command parameters from command definition.
|
||||
Args:
|
||||
cmd_name: Name of the command
|
||||
qubits: Ordered list of qubits command applies to
|
||||
|
||||
Raises:
|
||||
PulseError: If command for qubits is not available
|
||||
"""
|
||||
qubits = _to_qubit_tuple(qubits)
|
||||
if self.has(cmd_name, qubits):
|
||||
schedule = self._cmd_dict[cmd_name][qubits]
|
||||
return schedule.parameters
|
||||
|
||||
else:
|
||||
raise PulseError('Command {0} for qubits {1} is not present '
|
||||
'in CmdDef'.format(cmd_name, qubits))
|
||||
|
||||
def pop(self, cmd_name: str, qubits: Union[int, Iterable[int]],
|
||||
**params: Dict[str, Union[float, complex]]) -> Schedule:
|
||||
"""Pop command from command definition.
|
||||
|
||||
Args:
|
||||
cmd_name: Name of the command
|
||||
qubits: Ordered list of qubits command applies to
|
||||
**params: Command parameters to be used to generate schedule
|
||||
|
||||
Raises:
|
||||
PulseError: If command for qubits is not available
|
||||
"""
|
||||
qubits = _to_qubit_tuple(qubits)
|
||||
if self.has(cmd_name, qubits):
|
||||
cmd_dict = self._cmd_dict[cmd_name]
|
||||
schedule = cmd_dict.pop(qubits)
|
||||
|
||||
if isinstance(schedule, ParameterizedSchedule):
|
||||
return schedule.bind_parameters(**params)
|
||||
|
||||
return schedule
|
||||
|
||||
else:
|
||||
raise PulseError('Command {0} for qubits {1} is not present '
|
||||
'in CmdDef'.format(cmd_name, qubits))
|
||||
|
||||
def cmds(self) -> List[str]:
|
||||
"""Return all command names available in CmdDef."""
|
||||
|
||||
return list(self._cmd_dict.keys())
|
||||
|
||||
def cmd_qubits(self, cmd_name: str) -> List[Tuple[int]]:
|
||||
"""Get all qubit orderings this command exists for."""
|
||||
if cmd_name in self._cmd_dict:
|
||||
return list(sorted(self._cmd_dict[cmd_name].keys()))
|
||||
|
||||
raise PulseError('Command %s does not exist in CmdDef.' % cmd_name)
|
||||
|
||||
def __repr__(self):
|
||||
return repr(self._cmd_dict)
|
||||
|
|
@ -23,7 +23,7 @@ class SamplePulse(Command):
|
|||
"""Create new sample pulse command.
|
||||
|
||||
Args:
|
||||
samples (ndarray): Complex array of pulse envelope.
|
||||
samples (Union[np.ndarray, List[Complex]]): Complex array of pulse envelope.
|
||||
name (str): Unique name to identify the pulse.
|
||||
Raises:
|
||||
PulseError: when pulse envelope amplitude exceeds 1.
|
||||
|
|
@ -33,7 +33,7 @@ class SamplePulse(Command):
|
|||
if np.any(np.abs(samples) > 1):
|
||||
raise PulseError('Absolute value of pulse envelope amplitude exceeds 1.')
|
||||
|
||||
self._samples = samples
|
||||
self._samples = np.asarray(samples, dtype=np.complex_)
|
||||
|
||||
@property
|
||||
def samples(self):
|
||||
|
|
|
|||
|
|
@ -121,7 +121,7 @@ def _fix_gaussian_width(gaussian_samples, amp: float, center: float, sigma: floa
|
|||
gaussian_samples -= zero_offset
|
||||
amp_scale_factor = 1.
|
||||
if rescale_amp:
|
||||
amp_scale_factor = amp/(amp-zero_offset)
|
||||
amp_scale_factor = amp/(amp-zero_offset) if amp-zero_offset != 0 else 1.
|
||||
gaussian_samples *= amp_scale_factor
|
||||
|
||||
if ret_scale_factor:
|
||||
|
|
|
|||
|
|
@ -11,15 +11,12 @@
|
|||
# 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.
|
||||
|
||||
# pylint: disable=missing-return-doc,cyclic-import
|
||||
|
||||
"""
|
||||
Schedule.
|
||||
"""
|
||||
import itertools
|
||||
import logging
|
||||
from typing import List, Tuple, Iterable, Union
|
||||
from typing import List, Tuple, Iterable, Union, Dict
|
||||
|
||||
from qiskit.pulse import ops
|
||||
from .channels import Channel
|
||||
|
|
@ -29,6 +26,8 @@ from .exceptions import PulseError
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# pylint: disable=missing-return-doc,cyclic-import
|
||||
|
||||
|
||||
class Schedule(ScheduleComponent):
|
||||
"""Schedule of `ScheduleComponent`s. The composite node of a schedule tree."""
|
||||
|
|
@ -210,3 +209,59 @@ class Schedule(ScheduleComponent):
|
|||
if len(instructions) > 50:
|
||||
return res + ', ...)'
|
||||
return res + ')'
|
||||
|
||||
|
||||
class ParameterizedSchedule:
|
||||
"""Temporary parameterized schedule class.
|
||||
|
||||
This should not be returned to users as it is currently only a helper class.
|
||||
|
||||
This class is takes an input command definition that accepts
|
||||
a set of parameters. Calling `bind` on the class will return a `Schedule`.
|
||||
|
||||
# TODO: In the near future this will be replaced with proper incorporation of parameters
|
||||
into the `Schedule` class.
|
||||
"""
|
||||
|
||||
def __init__(self, *schedules, parameters=None, name=None):
|
||||
full_schedules = []
|
||||
parameterized = []
|
||||
parameters = parameters or []
|
||||
self.name = name or ''
|
||||
# partition schedules into callable and schedules
|
||||
for schedule in schedules:
|
||||
if isinstance(schedule, ParameterizedSchedule):
|
||||
parameterized.append(schedule)
|
||||
parameters += schedule.parameters
|
||||
elif callable(schedule):
|
||||
parameterized.append(schedule)
|
||||
elif isinstance(schedule, Schedule):
|
||||
full_schedules.append(schedule)
|
||||
else:
|
||||
raise PulseError('Input type: {0} not supported'.format(type(schedule)))
|
||||
|
||||
self._parameterized = tuple(parameterized)
|
||||
self._schedules = tuple(full_schedules)
|
||||
self._parameters = tuple(sorted(parameters))
|
||||
|
||||
@property
|
||||
def parameters(self) -> Tuple[str]:
|
||||
"""Schedule parameters."""
|
||||
return self._parameters
|
||||
|
||||
def bind_parameters(self, *args: List[float], **kwargs: Dict[str, float]) -> Schedule:
|
||||
"""Generate the Schedule from params to evaluate command expressions"""
|
||||
bound_schedule = Schedule(name=self.name)
|
||||
schedules = list(self._schedules)
|
||||
for param_sched in self._parameterized:
|
||||
# recursively call until based callable is reached
|
||||
schedules.append(param_sched(*args, **kwargs))
|
||||
|
||||
# construct evaluated schedules
|
||||
for sched in schedules:
|
||||
bound_schedule |= sched
|
||||
|
||||
return bound_schedule
|
||||
|
||||
def __call__(self, *args: List[float], **kwargs: Dict[str, float]) -> Schedule:
|
||||
return self.bind_parameters(*args, **kwargs)
|
||||
|
|
|
|||
|
|
@ -19,7 +19,8 @@ from .models.base import (QobjInstruction, QobjExperimentHeader, QobjExperimentC
|
|||
|
||||
from .models.pulse import (PulseQobjInstruction, PulseQobjExperimentConfig,
|
||||
PulseQobjExperiment, PulseQobjConfig,
|
||||
QobjMeasurementOption, QobjPulseLibrary)
|
||||
QobjMeasurementOption, PulseLibraryItem,
|
||||
PulseLibraryItemSchema, PulseQobjInstructionSchema)
|
||||
|
||||
from .models.qasm import (QasmQobjInstruction, QasmQobjExperimentConfig,
|
||||
QasmQobjExperiment, QasmQobjConfig)
|
||||
|
|
|
|||
|
|
@ -9,5 +9,5 @@
|
|||
Helper modules to convert qiskit frontend object to proper qobj model.
|
||||
"""
|
||||
|
||||
from .pulse_instruction import PulseQobjConverter
|
||||
from .pulse_instruction import InstructionToQobjConverter, QobjToInstructionConverter
|
||||
from .lo_config import LoConfigConverter
|
||||
|
|
|
|||
|
|
@ -6,77 +6,67 @@
|
|||
# the LICENSE.txt file in the root directory of this source tree.
|
||||
|
||||
"""Helper class used to convert a pulse instruction into PulseQobjInstruction."""
|
||||
import re
|
||||
import math
|
||||
|
||||
import functools
|
||||
from sympy.parsing.sympy_parser import (parse_expr, standard_transformations,
|
||||
implicit_multiplication_application,
|
||||
function_exponentiation)
|
||||
from sympy import Symbol
|
||||
|
||||
from qiskit.pulse import commands
|
||||
from qiskit.pulse import commands, channels, Schedule
|
||||
from qiskit.pulse.schedule import ParameterizedSchedule
|
||||
from qiskit.pulse.exceptions import PulseError
|
||||
from qiskit.qobj import QobjMeasurementOption
|
||||
from qiskit import QiskitError
|
||||
|
||||
|
||||
class ConversionMethodBinder:
|
||||
"""Instruction conversion method registrar."""
|
||||
"""Conversion method registrar."""
|
||||
def __init__(self):
|
||||
"""Acts as method registration decorator and tracker for instruction conversion methods."""
|
||||
"""Acts as method registration decorator and tracker for conversion methods."""
|
||||
self._bound_instructions = {}
|
||||
|
||||
def __call__(self, type_instruction):
|
||||
def __call__(self, bound):
|
||||
""" Converter decorator method.
|
||||
|
||||
Pulse instruction converter is defined for each instruction type,
|
||||
and this decorator binds converter function to valid instruction type.
|
||||
Converter is defined for object to be converted matched on hash
|
||||
|
||||
Args:
|
||||
type_instruction (Instruction): valid pulse instruction class to the converter.
|
||||
bound (Hashable): Hashable object to bind to the converter.
|
||||
|
||||
"""
|
||||
# pylint: disable=missing-return-doc, missing-return-type-doc
|
||||
|
||||
def _apply_converter(converter):
|
||||
"""Return decorated converter function."""
|
||||
|
||||
@functools.wraps(converter)
|
||||
def _call_valid_converter(self, shift, instruction):
|
||||
"""Return a dictionary for to be used to construct a qobj
|
||||
if the given instruction matches the
|
||||
bound instruction type supplied to the function,
|
||||
otherwise return None."""
|
||||
if isinstance(instruction, type_instruction):
|
||||
return converter(self, shift, instruction)
|
||||
else:
|
||||
raise PulseError('Supplied instruction {0} '
|
||||
'is not of type {1}.'.format(instruction, type_instruction))
|
||||
|
||||
# Track conversion methods for class.
|
||||
self._bound_instructions[type_instruction] = _call_valid_converter
|
||||
return _call_valid_converter
|
||||
self._bound_instructions[bound] = converter
|
||||
return converter
|
||||
|
||||
return _apply_converter
|
||||
|
||||
def get_bound_method(self, instruction):
|
||||
"""Get conversion method for instruction."""
|
||||
def get_bound_method(self, bound):
|
||||
"""Get conversion method for bound object."""
|
||||
try:
|
||||
return self._bound_instructions[type(instruction)]
|
||||
return self._bound_instructions[bound]
|
||||
except KeyError:
|
||||
raise PulseError('Qobj conversion method for %s is not found.' % instruction)
|
||||
raise PulseError('Bound method for %s is not found.' % bound)
|
||||
|
||||
|
||||
class PulseQobjConverter:
|
||||
"""
|
||||
This class exists for separating entity of pulse instructions and qobj instruction,
|
||||
and provides some alleviation of the complexity of the assembler.
|
||||
class InstructionToQobjConverter:
|
||||
"""Converts pulse Instructions to Qobj models.
|
||||
|
||||
Converter is constructed with qobj model and some experimental configuration,
|
||||
Converter is constructed with qobj model and experimental configuration,
|
||||
and returns proper qobj instruction to each backend.
|
||||
|
||||
Pulse instruction and its qobj are strongly depends on the design of backend,
|
||||
and third party providers can be easily add their custom pulse instruction by
|
||||
providing custom converter inherit from this.
|
||||
Third party providers can be add their own custom pulse instructions by
|
||||
providing custom converter methods.
|
||||
|
||||
|
||||
To create custom converter for custom instruction
|
||||
To create a custom converter for custom instruction
|
||||
```
|
||||
class CustomConverter(PulseQobjConverter):
|
||||
class CustomConverter(InstructionToQobjConverter):
|
||||
|
||||
@bind_instruction(CustomInstruction)
|
||||
def convert_custom_command(self, shift, instruction):
|
||||
|
|
@ -108,7 +98,7 @@ class PulseQobjConverter:
|
|||
|
||||
def __call__(self, shift, instruction):
|
||||
|
||||
method = self.bind_instruction.get_bound_method(instruction)
|
||||
method = self.bind_instruction.get_bound_method(type(instruction))
|
||||
return method(self, shift, instruction)
|
||||
|
||||
@bind_instruction(commands.AcquireInstruction)
|
||||
|
|
@ -226,3 +216,287 @@ class PulseQobjConverter:
|
|||
'type': instruction.type
|
||||
}
|
||||
return self._qobj_model(**command_dict)
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
|
||||
# get math operations valid in python. Presumably these are valid in sympy
|
||||
_math_ops = [math_op for math_op in math.__dict__ if not math_op.startswith('__')]
|
||||
# only allow valid math ops
|
||||
_math_ops_regex = r"(" + ")|(".join(_math_ops) + ")"
|
||||
# match consecutive alphanumeric, and single consecutive math ops +-/.()
|
||||
# and multiple * for exponentiation
|
||||
_allowedchars = re.compile(r'([+\/\-\(\)\.])?([\sa-zA-Z\d]+[+\/\-\(\)\.]?\*{0,2})*')
|
||||
# match any sequence of chars and numbers
|
||||
_expr_regex = r'([a-zA-Z]+\d*)'
|
||||
# and valid params
|
||||
_param_regex = r'(P\d+)'
|
||||
# only valid sequences are P# for parameters and valid math operations above
|
||||
_valid_sub_expr = re.compile(_param_regex+'|'+_math_ops_regex)
|
||||
# pylint: enable=invalid-name
|
||||
|
||||
|
||||
def _is_math_expr_safe(expr):
|
||||
r"""Verify mathematical expression is sanitized.
|
||||
|
||||
Only allow strings of form 'P\d+' and operations from `math`.
|
||||
Allowed chars are [a-zA-Z]. Allowed math operators are '+*/().'
|
||||
where only '*' are allowed to be consecutive.
|
||||
|
||||
Args:
|
||||
expr (str): Expression to sanitize
|
||||
|
||||
Returns:
|
||||
bool: Whether the string is safe to parse math from
|
||||
|
||||
Raise:
|
||||
QiskitError: If math expression is not sanitized
|
||||
"""
|
||||
|
||||
only_allowed_chars = _allowedchars.match(expr)
|
||||
if not only_allowed_chars:
|
||||
return False
|
||||
elif not only_allowed_chars.group(0) == expr:
|
||||
return False
|
||||
|
||||
sub_expressions = re.findall(_expr_regex, expr)
|
||||
if not all([_valid_sub_expr.match(sub_exp) for sub_exp in sub_expressions]):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _parse_string_expr(expr): # pylint: disable=missing-return-type-doc
|
||||
"""Parse a mathematical string expression and extract free parameters.
|
||||
|
||||
Args:
|
||||
expr (str): String expression to parse
|
||||
|
||||
Returns:
|
||||
(Callable, Tuple[str]): Returns a callable function and tuple of string symbols
|
||||
|
||||
Raises:
|
||||
QiskitError: If expression is not safe
|
||||
"""
|
||||
# remove these strings from expression and hope sympy knows these math expressions
|
||||
# these are effectively reserved keywords
|
||||
subs = [('numpy.', ''), ('np.', ''), ('math.', '')]
|
||||
for match, sub in subs:
|
||||
expr = expr.replace(match, sub)
|
||||
if not _is_math_expr_safe(expr):
|
||||
raise QiskitError('Expression: "%s" is not safe to evaluate.' % expr)
|
||||
params = sorted(re.findall(_param_regex, expr))
|
||||
local_dict = {param: Symbol(param) for param in params}
|
||||
symbols = list(local_dict.keys())
|
||||
transformations = (standard_transformations + (implicit_multiplication_application,) +
|
||||
(function_exponentiation,))
|
||||
|
||||
parsed_expr = parse_expr(expr, local_dict=local_dict, transformations=transformations)
|
||||
|
||||
def parsed_fun(*args, **kwargs):
|
||||
subs = {}
|
||||
matched_params = []
|
||||
if args:
|
||||
subs.update({symbols[i]: arg for i, arg in enumerate(args)})
|
||||
matched_params += list(params[i] for i in range(len(args)))
|
||||
elif kwargs:
|
||||
subs.update({local_dict[key]: value for key, value in kwargs.items()
|
||||
if key in local_dict})
|
||||
matched_params += list(key for key in kwargs if key in params)
|
||||
|
||||
if not set(matched_params).issuperset(set(params)):
|
||||
raise PulseError('Supplied params ({args}, {kwargs}) do not match '
|
||||
'{params}'.format(args=args, kwargs=kwargs, params=params))
|
||||
|
||||
return complex(parsed_expr.evalf(subs=subs))
|
||||
return parsed_fun, params
|
||||
|
||||
|
||||
class QobjToInstructionConverter:
|
||||
"""Converts Qobj models to pulse Instructions
|
||||
"""
|
||||
# pylint: disable=invalid-name,missing-return-type-doc
|
||||
# class level tracking of conversion methods
|
||||
bind_name = ConversionMethodBinder()
|
||||
chan_regex = re.compile(r'([a-zA-Z]+)(\d+)')
|
||||
|
||||
def __init__(self, pulse_library, buffer=0, **run_config):
|
||||
"""Create new converter.
|
||||
|
||||
Args:
|
||||
pulse_library (List[PulseLibraryItem]): Pulse library to be used in conversion
|
||||
buffer (int): Channel buffer
|
||||
run_config (dict): experimental configuration.
|
||||
"""
|
||||
self.buffer = buffer
|
||||
self._run_config = run_config
|
||||
|
||||
# bind pulses to conversion methods
|
||||
for pulse in pulse_library:
|
||||
self.bind_pulse(pulse)
|
||||
|
||||
def __call__(self, instruction):
|
||||
|
||||
method = self.bind_name.get_bound_method(instruction.name)
|
||||
return method(self, instruction)
|
||||
|
||||
def get_channel(self, channel):
|
||||
"""Parse and retrieve channel from ch string.
|
||||
|
||||
Args:
|
||||
channel (str): Channel to match
|
||||
|
||||
Returns:
|
||||
(Channel, int): Matched channel
|
||||
|
||||
Raises:
|
||||
PulseError: Is raised if valid channel is not matched
|
||||
"""
|
||||
match = self.chan_regex.match(channel)
|
||||
if match:
|
||||
prefix, index = match.group(1), int(match.group(2))
|
||||
|
||||
if prefix == channels.DriveChannel.prefix:
|
||||
return channels.DriveChannel(index, buffer=self.buffer)
|
||||
elif prefix == channels.MeasureChannel.prefix:
|
||||
return channels.MeasureChannel(index, buffer=self.buffer)
|
||||
elif prefix == channels.ControlChannel.prefix:
|
||||
return channels.ControlChannel(index, buffer=self.buffer)
|
||||
|
||||
raise PulseError('Channel %s is not valid' % channel)
|
||||
|
||||
@bind_name('acquire')
|
||||
def convert_acquire(self, instruction):
|
||||
"""Return converted `AcquireInstruction`.
|
||||
|
||||
Args:
|
||||
instruction (PulseQobjInstruction): acquire qobj
|
||||
Returns:
|
||||
Schedule: Converted and scheduled Instruction
|
||||
"""
|
||||
t0 = instruction.t0
|
||||
duration = instruction.duration
|
||||
qubits = instruction.qubits
|
||||
discriminators = (instruction.discriminators
|
||||
if hasattr(instruction, 'discriminators') else None)
|
||||
|
||||
kernels = (instruction.kernels
|
||||
if hasattr(instruction, 'kernels') else None)
|
||||
|
||||
mem_slots = instruction.memory_slot
|
||||
reg_slots = (instruction.register_slot
|
||||
if hasattr(instruction, 'register_slot') else None)
|
||||
|
||||
if not isinstance(discriminators, list):
|
||||
discriminators = [discriminators for _ in range(len(qubits))]
|
||||
|
||||
if not isinstance(kernels, list):
|
||||
kernels = [kernels for _ in range(len(qubits))]
|
||||
|
||||
schedule = Schedule()
|
||||
|
||||
for i, qubit in enumerate(qubits):
|
||||
kernel = kernels[i]
|
||||
if kernel:
|
||||
kernel = commands.Kernel(name=kernel.name,
|
||||
params=kernel.params)
|
||||
|
||||
discriminator = discriminators[i]
|
||||
if discriminator:
|
||||
discriminator = commands.Discriminator(name=discriminator.name,
|
||||
params=discriminator.params)
|
||||
|
||||
channel = channels.AcquireChannel(qubit, buffer=self.buffer)
|
||||
if reg_slots:
|
||||
register_slot = channels.RegisterSlot(reg_slots[i])
|
||||
else:
|
||||
register_slot = None
|
||||
memory_slot = channels.MemorySlot(mem_slots[i])
|
||||
|
||||
cmd = commands.Acquire(duration, discriminator=discriminator, kernel=kernel)
|
||||
schedule |= commands.AcquireInstruction(cmd, channel, memory_slot, register_slot) << t0
|
||||
|
||||
return schedule
|
||||
|
||||
@bind_name('fc')
|
||||
def convert_frame_change(self, instruction):
|
||||
"""Return converted `FrameChangeInstruction`.
|
||||
|
||||
Args:
|
||||
instruction (PulseQobjInstruction): frame change qobj
|
||||
Returns:
|
||||
Schedule: Converted and scheduled Instruction
|
||||
"""
|
||||
t0 = instruction.t0
|
||||
channel = self.get_channel(instruction.ch)
|
||||
phase = instruction.phase
|
||||
|
||||
# This is parameterized
|
||||
if isinstance(phase, str):
|
||||
phase_expr, params = _parse_string_expr(phase)
|
||||
|
||||
def gen_fc_sched(*args, **kwargs):
|
||||
phase = abs(phase_expr(*args, **kwargs))
|
||||
return commands.FrameChange(phase)(channel) << t0
|
||||
|
||||
return ParameterizedSchedule(gen_fc_sched, parameters=params)
|
||||
|
||||
return commands.FrameChange(phase)(channel) << t0
|
||||
|
||||
@bind_name('pv')
|
||||
def convert_persistent_value(self, instruction):
|
||||
"""Return converted `PersistentValueInstruction`.
|
||||
|
||||
Args:
|
||||
instruction (PulseQobjInstruction): persistent value qobj
|
||||
Returns:
|
||||
Schedule: Converted and scheduled Instruction
|
||||
"""
|
||||
t0 = instruction.t0
|
||||
channel = self.get_channel(instruction.ch)
|
||||
val = instruction.val
|
||||
|
||||
# This is parameterized
|
||||
if isinstance(val, str):
|
||||
val_expr, params = _parse_string_expr(val)
|
||||
|
||||
def gen_fc_sched(*args, **kwargs):
|
||||
val = complex(val_expr(*args, **kwargs))
|
||||
return commands.PersistentValue(val)(channel) << t0
|
||||
|
||||
return ParameterizedSchedule(gen_fc_sched, parameters=params)
|
||||
|
||||
return commands.PersistentValue(val)(channel) << t0
|
||||
|
||||
def bind_pulse(self, pulse):
|
||||
"""Bind the supplied pulse to a converter method by pulse name.
|
||||
|
||||
Args:
|
||||
pulse (PulseLibraryItem): Pulse to bind
|
||||
"""
|
||||
# pylint: disable=unused-variable
|
||||
pulse = commands.SamplePulse(pulse.samples, pulse.name)
|
||||
|
||||
@self.bind_name(pulse.name)
|
||||
def convert_named_drive(self, instruction):
|
||||
"""Return converted `PulseInstruction`.
|
||||
|
||||
Args:
|
||||
instruction (PulseQobjInstruction): pulse qobj
|
||||
Returns:
|
||||
Schedule: Converted and scheduled pulse
|
||||
"""
|
||||
t0 = instruction.t0
|
||||
channel = self.get_channel(instruction.ch)
|
||||
return pulse(channel) << t0
|
||||
|
||||
@bind_name('snapshot')
|
||||
def convert_snapshot(self, instruction):
|
||||
"""Return converted `Snapshot`.
|
||||
|
||||
Args:
|
||||
instruction (PulseQobjInstruction): snapshot qobj
|
||||
Returns:
|
||||
Schedule: Converted and scheduled Snapshot
|
||||
"""
|
||||
t0 = instruction.t0
|
||||
return commands.Snapshot(instruction.label, instruction.type) << t0
|
||||
|
|
|
|||
|
|
@ -11,7 +11,8 @@ from marshmallow.validate import Range, Regexp, Length, OneOf
|
|||
|
||||
from qiskit.qobj.utils import MeasReturnType
|
||||
from qiskit.validation import BaseSchema, bind_schema, BaseModel
|
||||
from qiskit.validation.fields import Integer, String, Number, Complex, List, Nested, DictParameters
|
||||
from qiskit.validation.fields import (Integer, String, Number, Complex, List,
|
||||
Nested, DictParameters, ByType)
|
||||
from .base import (QobjInstructionSchema, QobjExperimentConfigSchema, QobjExperimentSchema,
|
||||
QobjConfigSchema, QobjInstruction, QobjExperimentConfig,
|
||||
QobjExperiment, QobjConfig)
|
||||
|
|
@ -26,8 +27,8 @@ class QobjMeasurementOptionSchema(BaseSchema):
|
|||
required=True)
|
||||
|
||||
|
||||
class QobjPulseLibrarySchema(BaseSchema):
|
||||
"""Schema for QobjPulseLibrary."""
|
||||
class PulseLibraryItemSchema(BaseSchema):
|
||||
"""Schema for PulseLibraryItem."""
|
||||
|
||||
# Required properties.
|
||||
name = String(required=True)
|
||||
|
|
@ -44,8 +45,8 @@ class PulseQobjInstructionSchema(QobjInstructionSchema):
|
|||
# Optional properties.
|
||||
ch = String(validate=Regexp('[dum]([0-9])+'))
|
||||
conditional = Integer(validate=Range(min=0))
|
||||
phase = Number()
|
||||
val = Complex()
|
||||
val = ByType([Complex(), String()])
|
||||
phase = ByType([String(), Number()])
|
||||
duration = Integer(validate=Range(min=1))
|
||||
qubits = List(Integer(validate=Range(min=0)), validate=Length(min=1))
|
||||
memory_slot = List(Integer(validate=Range(min=0)), validate=Length(min=1))
|
||||
|
|
@ -83,7 +84,7 @@ class PulseQobjConfigSchema(QobjConfigSchema):
|
|||
meas_level = Integer(required=True, validate=Range(min=0, max=2))
|
||||
meas_return = String(required=True, validate=OneOf(choices=(MeasReturnType.AVERAGE,
|
||||
MeasReturnType.SINGLE)))
|
||||
pulse_library = Nested(QobjPulseLibrarySchema, required=True, many=True)
|
||||
pulse_library = Nested(PulseLibraryItemSchema, required=True, many=True)
|
||||
qubit_lo_freq = List(Number(validate=Range(min=0)), required=True)
|
||||
meas_lo_freq = List(Number(validate=Range(min=0)), required=True)
|
||||
|
||||
|
|
@ -110,12 +111,12 @@ class QobjMeasurementOption(BaseModel):
|
|||
super().__init__(**kwargs)
|
||||
|
||||
|
||||
@bind_schema(QobjPulseLibrarySchema)
|
||||
class QobjPulseLibrary(BaseModel):
|
||||
"""Model for QobjPulseLibrary.
|
||||
@bind_schema(PulseLibraryItemSchema)
|
||||
class PulseLibraryItem(BaseModel):
|
||||
"""Model for PulseLibraryItem.
|
||||
|
||||
Please note that this class only describes the required fields. For the
|
||||
full description of the model, please check ``QobjPulseLibrarySchema``.
|
||||
full description of the model, please check ``PulseLibraryItemSchema``.
|
||||
|
||||
Attributes:
|
||||
name (str): name of pulse
|
||||
|
|
@ -186,7 +187,7 @@ class PulseQobjConfig(QobjConfig):
|
|||
meas_level (int): a value represents the level of measurement.
|
||||
meas_lo_freq (list[float]): local oscillator frequency of measurement pulse.
|
||||
meas_return (str): a level of measurement information.
|
||||
pulse_library (list[qiskit.qobj.QobjPulseLibrary]): a pulse library.
|
||||
pulse_library (list[qiskit.qobj.PulseLibraryItem]): a pulse library.
|
||||
qubit_lo_freq (list[float]): local oscillator frequency of driving pulse.
|
||||
"""
|
||||
def __init__(self, meas_level, meas_return, pulse_library,
|
||||
|
|
|
|||
|
|
@ -115,13 +115,15 @@
|
|||
]
|
||||
},
|
||||
"val": {
|
||||
"items": {
|
||||
"OneOf": [
|
||||
{"items": {
|
||||
"type": "number"
|
||||
},
|
||||
"maxItems": 2,
|
||||
"minItems": 2,
|
||||
"type": "array"
|
||||
}
|
||||
"type": "array"},
|
||||
{"type": "string"}
|
||||
]}
|
||||
},
|
||||
"required": [
|
||||
"val",
|
||||
|
|
@ -138,7 +140,9 @@
|
|||
]
|
||||
},
|
||||
"phase": {
|
||||
"type": "number"
|
||||
"OneOf":[
|
||||
{"type": "number"},
|
||||
{"type": "string"}]
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
|
|
|
|||
|
|
@ -31,12 +31,13 @@ import time
|
|||
|
||||
from qiskit.result import Result
|
||||
from qiskit.providers import BaseBackend, BaseJob
|
||||
from qiskit.providers.models import (BackendProperties, PulseDefaults, UchannelLO,
|
||||
QasmBackendConfiguration, PulseBackendConfiguration)
|
||||
from qiskit.providers.models.backendconfiguration import GateConfig
|
||||
from qiskit.providers.models import (BackendProperties, GateConfig,
|
||||
QasmBackendConfiguration, PulseBackendConfiguration,
|
||||
PulseDefaults, Command, UchannelLO)
|
||||
from qiskit.qobj import (QasmQobj, QobjExperimentHeader, QobjHeader,
|
||||
QasmQobjInstruction, QasmQobjExperimentConfig,
|
||||
QasmQobjExperiment, QasmQobjConfig)
|
||||
QasmQobjExperiment, QasmQobjConfig, PulseLibraryItem,
|
||||
PulseQobjInstruction)
|
||||
from qiskit.providers.jobstatus import JobStatus
|
||||
from qiskit.providers.baseprovider import BaseProvider
|
||||
from qiskit.providers.exceptions import QiskitBackendNotFoundError
|
||||
|
|
@ -60,8 +61,9 @@ class FakeProvider(BaseProvider):
|
|||
# pylint: enable=no-member
|
||||
if not filtered_backends:
|
||||
raise QiskitBackendNotFoundError()
|
||||
else:
|
||||
backend = filtered_backends[0]
|
||||
|
||||
backend = filtered_backends[0]
|
||||
|
||||
return backend
|
||||
|
||||
def backends(self, name=None, **kwargs):
|
||||
|
|
@ -73,7 +75,7 @@ class FakeProvider(BaseProvider):
|
|||
FakeMelbourne(),
|
||||
FakeRueschlikon(),
|
||||
FakeTokyo(),
|
||||
FakeOpenPulse2Q()],
|
||||
FakeOpenPulse2Q()]
|
||||
super().__init__()
|
||||
|
||||
|
||||
|
|
@ -192,8 +194,23 @@ class FakeOpenPulse2Q(FakeBackend):
|
|||
qubit_freq_est=[4.9, 5.0],
|
||||
meas_freq_est=[6.5, 6.6],
|
||||
buffer=10,
|
||||
pulse_library=[],
|
||||
cmd_def=[]
|
||||
pulse_library=[PulseLibraryItem(name='test_pulse_1', samples=[0.j, 0.1j]),
|
||||
PulseLibraryItem(name='test_pulse_2', samples=[0.j, 0.1j, 1j])],
|
||||
cmd_def=[Command(name='u1', qubits=[0],
|
||||
sequence=[PulseQobjInstruction(name='fc', ch='d0',
|
||||
t0=0, phase='-P1*np.pi')]),
|
||||
Command(name='cx', qubits=[0, 1],
|
||||
sequence=[PulseQobjInstruction(name='test_pulse_1', ch='d0', t0=0),
|
||||
PulseQobjInstruction(name='test_pulse_2', ch='u0', t0=10),
|
||||
PulseQobjInstruction(name='pv', ch='d1',
|
||||
t0=2, val='cos(P2)'),
|
||||
PulseQobjInstruction(name='test_pulse_1', ch='d1', t0=20),
|
||||
PulseQobjInstruction(name='fc', ch='d1',
|
||||
t0=20, phase=2.1)]),
|
||||
Command(name='measure', qubits=[0],
|
||||
sequence=[PulseQobjInstruction(name='test_pulse_1', ch='m0', t0=0),
|
||||
PulseQobjInstruction(name='acquire', duration=10, t0=0,
|
||||
qubits=[0], memory_slot=[0])])]
|
||||
)
|
||||
|
||||
super().__init__(configuration)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,100 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2019, IBM.
|
||||
#
|
||||
# This source code is licensed under the Apache License, Version 2.0 found in
|
||||
# the LICENSE.txt file in the root directory of this source tree.
|
||||
|
||||
"""Test for the CmdDef object."""
|
||||
|
||||
import numpy as np
|
||||
|
||||
from qiskit.test import QiskitTestCase
|
||||
from qiskit.test.mock import FakeProvider
|
||||
from qiskit.qobj.converters import QobjToInstructionConverter
|
||||
from qiskit.qobj import PulseQobjInstruction
|
||||
from qiskit.pulse import (CmdDef, SamplePulse, Schedule, DeviceSpecification,
|
||||
PulseError, PersistentValue)
|
||||
|
||||
|
||||
class TestCmdDef(QiskitTestCase):
|
||||
"""Test CmdDef methods."""
|
||||
|
||||
def setUp(self):
|
||||
self.provider = FakeProvider()
|
||||
self.backend = self.provider.get_backend('fake_openpulse_2q')
|
||||
self.device = DeviceSpecification.create_from(self.backend)
|
||||
|
||||
def test_get_backend(self):
|
||||
"""Test that backend is fetchable with cmd def present."""
|
||||
|
||||
def test_init(self):
|
||||
"""Test `init`, `has`."""
|
||||
sched = Schedule()
|
||||
sched.append(SamplePulse(np.ones(5))(self.device.q[0].drive))
|
||||
cmd_def = CmdDef({('tmp', 0): sched})
|
||||
self.assertTrue(cmd_def.has('tmp', 0))
|
||||
|
||||
def test_add(self):
|
||||
"""Test `add`, `has`, `get`, `cmdss`."""
|
||||
sched = Schedule()
|
||||
sched.append(SamplePulse(np.ones(5))(self.device.q[0].drive))
|
||||
cmd_def = CmdDef()
|
||||
cmd_def.add('tmp', 1, sched)
|
||||
cmd_def.add('tmp', 0, sched)
|
||||
self.assertEqual(sched.instructions, cmd_def.get('tmp', (0,)).instructions)
|
||||
|
||||
self.assertIn('tmp', cmd_def.cmds())
|
||||
self.assertEqual(cmd_def.cmd_qubits('tmp'), [(0,), (1,)])
|
||||
|
||||
def test_pop(self):
|
||||
"""Test pop with default."""
|
||||
sched = Schedule()
|
||||
sched.append(SamplePulse(np.ones(5))(self.device.q[0].drive))
|
||||
cmd_def = CmdDef()
|
||||
cmd_def.add('tmp', 0, sched)
|
||||
cmd_def.pop('tmp', 0)
|
||||
self.assertFalse(cmd_def.has('tmp', 0))
|
||||
|
||||
with self.assertRaises(PulseError):
|
||||
cmd_def.pop('not_there', (0,))
|
||||
|
||||
def test_repr(self):
|
||||
"""Test repr."""
|
||||
sched = Schedule()
|
||||
sched.append(SamplePulse(np.ones(5))(self.device.q[0].drive))
|
||||
cmd_def = CmdDef({('tmp', 0): sched})
|
||||
repr(cmd_def)
|
||||
|
||||
def test_parameterized_schedule(self):
|
||||
"""Test building parameterized schedule."""
|
||||
cmd_def = CmdDef()
|
||||
converter = QobjToInstructionConverter([], buffer=0)
|
||||
qobj = PulseQobjInstruction(name='pv', ch='u1', t0=10, val='P2*cos(np.pi*P1)')
|
||||
converted_instruction = converter(qobj)
|
||||
|
||||
cmd_def.add('pv_test', 0, converted_instruction)
|
||||
self.assertEqual(cmd_def.get_parameters('pv_test', 0), ('P1', 'P2'))
|
||||
|
||||
sched = cmd_def.get('pv_test', 0, P1='0', P2=-1)
|
||||
self.assertEqual(sched.instructions[0][-1].command.value, -1)
|
||||
|
||||
def test_build_cmd_def(self):
|
||||
"""Test building of parameterized cmd_def from defaults."""
|
||||
defaults = self.backend.defaults()
|
||||
cmd_def = defaults.build_cmd_def()
|
||||
|
||||
cx_pv = cmd_def.get('cx', (0, 1), P2=0)
|
||||
pv_found = False
|
||||
for _, instr in cx_pv.instructions:
|
||||
cmd = instr.command
|
||||
if isinstance(cmd, PersistentValue):
|
||||
self.assertEqual(cmd.value, 1)
|
||||
pv_found = True
|
||||
self.assertTrue(pv_found)
|
||||
|
||||
self.assertEqual(cmd_def.get_parameters('u1', 0), ('P1',))
|
||||
|
||||
u1_minus_pi = cmd_def.get('u1', 0, P1=1)
|
||||
fc_cmd = u1_minus_pi.instructions[0][-1].command
|
||||
self.assertEqual(fc_cmd.phase, np.pi)
|
||||
|
|
@ -12,15 +12,20 @@
|
|||
import numpy as np
|
||||
|
||||
from qiskit.test import QiskitTestCase
|
||||
from qiskit.qobj import PulseQobjInstruction, PulseQobjExperimentConfig
|
||||
from qiskit.qobj.converters import PulseQobjConverter, LoConfigConverter
|
||||
from qiskit.pulse.commands import SamplePulse, FrameChange, PersistentValue, Snapshot, Acquire
|
||||
from qiskit.qobj import (PulseQobjInstruction, PulseQobjExperimentConfig, PulseLibraryItem,
|
||||
QobjMeasurementOption)
|
||||
from qiskit.qobj.converters import (InstructionToQobjConverter, QobjToInstructionConverter,
|
||||
LoConfigConverter)
|
||||
from qiskit.qobj.converters.pulse_instruction import _is_math_expr_safe
|
||||
from qiskit.pulse.commands import (SamplePulse, FrameChange, PersistentValue, Snapshot, Acquire,
|
||||
Discriminator, Kernel)
|
||||
from qiskit.pulse.channels import (DeviceSpecification, Qubit, AcquireChannel, DriveChannel,
|
||||
MeasureChannel, RegisterSlot, MemorySlot)
|
||||
ControlChannel, MeasureChannel, RegisterSlot, MemorySlot,)
|
||||
from qiskit.pulse.schedule import ParameterizedSchedule
|
||||
from qiskit.pulse import LoConfig
|
||||
|
||||
|
||||
class TestInstructionConverter(QiskitTestCase):
|
||||
class TestInstructionToQobjConverter(QiskitTestCase):
|
||||
"""Pulse converter tests."""
|
||||
|
||||
def setUp(self):
|
||||
|
|
@ -38,7 +43,7 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
|
||||
def test_drive_instruction(self):
|
||||
"""Test converted qobj from PulseInstruction."""
|
||||
converter = PulseQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
converter = InstructionToQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
command = SamplePulse(np.arange(0, 0.01), name='linear')
|
||||
instruction = command(self.device.q[0].drive)
|
||||
|
||||
|
|
@ -52,7 +57,7 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
|
||||
def test_frame_change(self):
|
||||
"""Test converted qobj from FrameChangeInstruction."""
|
||||
converter = PulseQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
converter = InstructionToQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
command = FrameChange(phase=0.1)
|
||||
instruction = command(self.device.q[0].drive)
|
||||
|
||||
|
|
@ -67,7 +72,7 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
|
||||
def test_persistent_value(self):
|
||||
"""Test converted qobj from PersistentValueInstruction."""
|
||||
converter = PulseQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
converter = InstructionToQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
command = PersistentValue(value=0.1j)
|
||||
instruction = command(self.device.q[0].drive)
|
||||
|
||||
|
|
@ -82,7 +87,7 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
|
||||
def test_acquire(self):
|
||||
"""Test converted qobj from AcquireInstruction."""
|
||||
converter = PulseQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
converter = InstructionToQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
command = Acquire(duration=10)
|
||||
instruction = command(self.device.q, self.device.mem, self.device.c)
|
||||
|
||||
|
|
@ -99,7 +104,7 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
|
||||
def test_snapshot(self):
|
||||
"""Test converted qobj from SnapShot."""
|
||||
converter = PulseQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
converter = InstructionToQobjConverter(PulseQobjInstruction, meas_level=2)
|
||||
instruction = Snapshot(name='label', snap_type='type')
|
||||
|
||||
valid_qobj = PulseQobjInstruction(
|
||||
|
|
@ -112,6 +117,138 @@ class TestInstructionConverter(QiskitTestCase):
|
|||
self.assertEqual(converter(0, instruction), valid_qobj)
|
||||
|
||||
|
||||
class TestQobjToInstructionConverter(QiskitTestCase):
|
||||
"""Pulse converter tests."""
|
||||
|
||||
def setUp(self):
|
||||
self.linear = SamplePulse(np.arange(0, 0.01), name='linear')
|
||||
self.pulse_library = [PulseLibraryItem(name=self.linear.name,
|
||||
samples=self.linear.samples.tolist())]
|
||||
|
||||
self.converter = QobjToInstructionConverter(self.pulse_library, buffer=0)
|
||||
|
||||
self.device = DeviceSpecification(
|
||||
qubits=[
|
||||
Qubit(0, DriveChannel(0), MeasureChannel(0), AcquireChannel(0))
|
||||
],
|
||||
registers=[
|
||||
RegisterSlot(0)
|
||||
],
|
||||
mem_slots=[
|
||||
MemorySlot(0)
|
||||
]
|
||||
)
|
||||
|
||||
def test_drive_instruction(self):
|
||||
"""Test converted qobj from PulseInstruction."""
|
||||
cmd = self.linear
|
||||
instruction = cmd(DriveChannel(0)) << 10
|
||||
|
||||
qobj = PulseQobjInstruction(name='linear', ch='d0', t0=10)
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertEqual(converted_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(converted_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_frame_change(self):
|
||||
"""Test converted qobj from FrameChangeInstruction."""
|
||||
cmd = FrameChange(phase=0.1)
|
||||
instruction = cmd(MeasureChannel(0))
|
||||
|
||||
qobj = PulseQobjInstruction(name='fc', ch='m0', t0=0, phase=0.1)
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertEqual(converted_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(converted_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_persistent_value(self):
|
||||
"""Test converted qobj from PersistentValueInstruction."""
|
||||
cmd = PersistentValue(value=0.1j)
|
||||
instruction = cmd(ControlChannel(1))
|
||||
|
||||
qobj = PulseQobjInstruction(name='pv', ch='u1', t0=0, val=0.1j)
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertEqual(converted_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(converted_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_acquire(self):
|
||||
"""Test converted qobj from AcquireInstruction."""
|
||||
cmd = Acquire(10, Discriminator(name='test_disc', params={'test_params': 1.0}),
|
||||
Kernel(name='test_kern', params={'test_params': 'test'}))
|
||||
instruction = cmd(self.device.q, self.device.mem, self.device.c)
|
||||
|
||||
qobj = PulseQobjInstruction(name='acquire', t0=0, duration=10, qubits=[0],
|
||||
memory_slot=[0], register_slot=[0],
|
||||
kernels=[QobjMeasurementOption(
|
||||
name='test_kern', params={'test_params': 'test'})],
|
||||
discriminators=[QobjMeasurementOption(
|
||||
name='test_disc', params={'test_params': 1.0})])
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertEqual(converted_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(converted_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_snapshot(self):
|
||||
"""Test converted qobj from SnapShot."""
|
||||
cmd = Snapshot(name='label', snap_type='type')
|
||||
instruction = cmd << 10
|
||||
|
||||
qobj = PulseQobjInstruction(name='snapshot', t0=10, label='label', type='type')
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertEqual(converted_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(converted_instruction.instructions[0][-1], cmd)
|
||||
|
||||
def test_parameterized_frame_change(self):
|
||||
"""Test converted qobj from FrameChangeInstruction."""
|
||||
cmd = FrameChange(phase=4.)
|
||||
instruction = cmd(MeasureChannel(0)) << 10
|
||||
|
||||
qobj = PulseQobjInstruction(name='fc', ch='m0', t0=10, phase='P1**2')
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertIsInstance(converted_instruction, ParameterizedSchedule)
|
||||
|
||||
evaluated_instruction = converted_instruction.bind_parameters(2.)
|
||||
|
||||
self.assertEqual(evaluated_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(evaluated_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_parameterized_persistent_value(self):
|
||||
"""Test converted qobj from PersistentValueInstruction."""
|
||||
cmd = PersistentValue(value=0.5+0.j)
|
||||
instruction = cmd(ControlChannel(1)) << 10
|
||||
|
||||
qobj = PulseQobjInstruction(name='pv', ch='u1', t0=10, val='P1*cos(np.pi*P2)')
|
||||
converted_instruction = self.converter(qobj)
|
||||
|
||||
self.assertIsInstance(converted_instruction, ParameterizedSchedule)
|
||||
|
||||
evaluated_instruction = converted_instruction.bind_parameters(P1=0.5, P2=0.)
|
||||
|
||||
self.assertEqual(evaluated_instruction.timeslots, instruction.timeslots)
|
||||
self.assertEqual(evaluated_instruction.instructions[0][-1].command, cmd)
|
||||
|
||||
def test_expression_sanitizer(self):
|
||||
"""Test math expression sanitization."""
|
||||
|
||||
self.assertFalse(_is_math_expr_safe('INSERT INTO students VALUES (?,?)'))
|
||||
self.assertFalse(_is_math_expr_safe('import math'))
|
||||
self.assertFalse(_is_math_expr_safe('complex'))
|
||||
self.assertFalse(_is_math_expr_safe('2***2'))
|
||||
self.assertFalse(_is_math_expr_safe('avdfd*3'))
|
||||
self.assertFalse(_is_math_expr_safe('Cos(1+2)'))
|
||||
self.assertFalse(_is_math_expr_safe('hello_world'))
|
||||
self.assertFalse(_is_math_expr_safe('1_2'))
|
||||
self.assertFalse(_is_math_expr_safe('2+-2'))
|
||||
|
||||
self.assertTrue(_is_math_expr_safe('1+1*2*3.2+8*cos(1)**2'))
|
||||
self.assertTrue(_is_math_expr_safe('pi*2'))
|
||||
self.assertTrue(_is_math_expr_safe('-P1*cos(P2)'))
|
||||
self.assertTrue(_is_math_expr_safe('-P1*P2*P3'))
|
||||
|
||||
|
||||
class TestLoConverter(QiskitTestCase):
|
||||
"""LO converter tests."""
|
||||
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ from qiskit.providers.basicaer import basicaerjob
|
|||
from qiskit.qobj import (QasmQobj, PulseQobj, QobjHeader,
|
||||
PulseQobjInstruction, PulseQobjExperiment,
|
||||
PulseQobjConfig, QobjMeasurementOption,
|
||||
QobjPulseLibrary, QasmQobjInstruction,
|
||||
PulseLibraryItem, QasmQobjInstruction,
|
||||
QasmQobjExperiment, QasmQobjConfig)
|
||||
from qiskit.qobj import validate_qobj_against_schema
|
||||
from qiskit.validation.jsonschema.exceptions import SchemaValidationError
|
||||
|
|
@ -146,7 +146,7 @@ class TestPulseQobj(QiskitTestCase):
|
|||
memory_slot_size=8192,
|
||||
meas_return='avg',
|
||||
pulse_library=[
|
||||
QobjPulseLibrary(name='pulse0',
|
||||
PulseLibraryItem(name='pulse0',
|
||||
samples=[0.0 + 0.0j,
|
||||
0.5 + 0.0j,
|
||||
0.0 + 0.0j])
|
||||
|
|
@ -221,7 +221,7 @@ class TestPulseQobj(QiskitTestCase):
|
|||
memory_slot_size=8192,
|
||||
meas_return='avg',
|
||||
pulse_library=[
|
||||
QobjPulseLibrary(name='pulse0', samples=[0.1 + 0.0j])
|
||||
PulseLibraryItem(name='pulse0', samples=[0.1 + 0.0j])
|
||||
],
|
||||
qubit_lo_freq=[4.9], meas_lo_freq=[6.9],
|
||||
rep_time=1000),
|
||||
|
|
@ -233,8 +233,8 @@ class TestPulseQobj(QiskitTestCase):
|
|||
'meas_lo_freq': [6.9],
|
||||
'rep_time': 1000},
|
||||
),
|
||||
QobjPulseLibrary: (
|
||||
QobjPulseLibrary(name='pulse0', samples=[0.1 + 0.0j]),
|
||||
PulseLibraryItem: (
|
||||
PulseLibraryItem(name='pulse0', samples=[0.1 + 0.0j]),
|
||||
{'name': 'pulse0', 'samples': [[0.1, 0.0]]}
|
||||
),
|
||||
PulseQobjExperiment: (
|
||||
|
|
|
|||
Loading…
Reference in New Issue