Build pulse sequences from circuits

src/braket/circuits/__init__.py

@@ -22,6 +22,7 @@
 from braket.circuits.angled_gate import AngledGate, DoubleAngledGate  # noqa: F401
 from braket.circuits.circuit import Circuit  # noqa: F401
 from braket.circuits.circuit_diagram import CircuitDiagram  # noqa: F401
+from braket.circuits.circuit_pulse_sequence import CircuitPulseSequenceBuilder  # noqa: F401
 from braket.circuits.compiler_directive import CompilerDirective  # noqa: F401
 from braket.circuits.free_parameter import FreeParameter  # noqa: F401
 from braket.circuits.free_parameter_expression import FreeParameterExpression  # noqa: F401

src/braket/circuits/circuit.py

@@ -15,13 +15,14 @@
 
 from collections.abc import Callable, Iterable
 from numbers import Number
-from typing import Any, Optional, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Optional, TypeVar, Union
 
 import numpy as np
 import oqpy
 from sympy import Expr
 
 from braket.circuits import compiler_directives
+from braket.circuits.circuit_pulse_sequence import CircuitPulseSequenceBuilder
 from braket.circuits.free_parameter import FreeParameter
 from braket.circuits.free_parameter_expression import FreeParameterExpression
 from braket.circuits.gate import Gate
@@ -63,6 +64,9 @@
 from braket.registers.qubit import QubitInput
 from braket.registers.qubit_set import QubitSet, QubitSetInput
 
+if TYPE_CHECKING:  # pragma: no cover
+    from braket.aws import AwsDevice
+
 SubroutineReturn = TypeVar(
     "SubroutineReturn", Iterable[Instruction], Instruction, ResultType, Iterable[ResultType]
 )
@@ -1098,6 +1102,26 @@ def diagram(self, circuit_diagram_class: type = UnicodeCircuitDiagram) -> str:
         """
         return circuit_diagram_class.build_diagram(self)
 
+    def pulse_sequence(
+        self,
+        device: AwsDevice,
+        gate_definitions: dict[tuple[Gate, QubitSet], PulseSequence] | None = None,
+        pulse_sequence_builder_class: type = CircuitPulseSequenceBuilder,
+    ) -> PulseSequence:
+        """Get the associated pulse sequence for the current circuit.
+
+        Args:
+            device (AwsDevice): an AWS device.
+            gate_definitions (dict[tuple[Gate, QubitSet], PulseSequence] | None): Additional
+                gate definitions.
+            pulse_sequence_builder_class (type): A `CircuitPulseSequenceBuilder` class that builds
+                the pulse sequence for this circuit. Default = `CircuitPulseSequenceBuilder`.
+
+        Returns:
+            PulseSequence: A PulseSequence corresponding to the full circuit.
+        """
+        return pulse_sequence_builder_class(device, gate_definitions).build_pulse_sequence(self)
+
     def to_ir(
         self,
         ir_type: IRType = IRType.JAQCD,

src/braket/circuits/circuit_pulse_sequence.py

@@ -0,0 +1,168 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+#     http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+from __future__ import annotations
+
+import warnings
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:  # pragma: no cover
+    from braket.aws.aws_device import AwsDevice
+
+import braket.circuits.circuit as cir
+from braket.circuits.gate import Gate
+from braket.circuits.gate_calibrations import GateCalibrations
+from braket.circuits.qubit_set import QubitSet
+from braket.circuits.result_type import ResultType
+from braket.parametric.free_parameter import FreeParameter
+from braket.pulse.frame import Frame
+from braket.pulse.pulse_sequence import PulseSequence
+
+
+class CircuitPulseSequenceBuilder:
+    """Builds a pulse sequence from circuits."""
+
+    def __init__(
+        self,
+        device: AwsDevice,
+        gate_definitions: dict[tuple[Gate, QubitSet], PulseSequence] | None = None,
+    ) -> None:
+        _validate_device(device)
+        gate_definitions = gate_definitions or {}
+
+        self._device = device
+        self._gate_calibrations = (
+            device.gate_calibrations
+            if device.gate_calibrations is not None
+            else GateCalibrations({})
+        )
+        self._gate_calibrations.pulse_sequences.update(gate_definitions)
+
+    def build_pulse_sequence(self, circuit: cir.Circuit) -> PulseSequence:
+        """
+        Build a PulseSequence corresponding to the full circuit.
+
+        Args:
+            circuit (Circuit): Circuit for which to build a diagram.
+
+        Returns:
+            PulseSequence: a pulse sequence created from all gates.
+        """
+
+        pulse_sequence = PulseSequence()
+        if not circuit.instructions:
+            return pulse_sequence
+
+        if circuit.parameters:
+            raise ValueError("All parameters must be assigned to draw the pulse sequence.")
+
+        for instruction in circuit.instructions:
+            gate = instruction.operator
+            qubits = instruction.target
+
+            gate_pulse_sequence = self._get_pulse_sequence(gate, qubits)
+
+            # FIXME: this creates a single cal block, "barrier;" in defcal could be either
+            # global or restricted to the defcal context
+            # Right now they are global
+            pulse_sequence += gate_pulse_sequence
+
+        # Result type columns
+        target_result_types = CircuitPulseSequenceBuilder._categorize_result_types(
+            circuit.result_types
+        )
+
+        for result_type in target_result_types:
+            pulse_sequence += result_type._to_pulse_sequence()
+
+        for qubit in circuit.qubits:
+            pulse_sequence.capture_v0(self._readout_frame(qubit))
+
+        return pulse_sequence
+
+    def _get_pulse_sequence(self, gate: Gate, qubit: QubitSet) -> PulseSequence:
+        parameters = gate.parameters if hasattr(gate, "parameters") else []
+        if isinstance(gate, Gate) and gate.name == "PulseGate":
+            gate_pulse_sequence = gate.pulse_sequence
+        elif (
+            gate_pulse_sequence := self._gate_calibrations.pulse_sequences.get((gate, qubit), None)
+        ) is None:
+            if (
+                not hasattr(gate, "parameters")
+                or (
+                    gate_pulse_sequence := self._find_parametric_gate_calibration(
+                        gate, qubit, len(gate.parameters)
+                    )
+                )
+                is None
+            ):
+                parameter_str = ", ".join(str(p) for p in parameters)
+                qubit_str = ", ".join(str(int(q)) for q in qubit)
+                raise ValueError(
+                    f"No pulse sequence for {gate.name}({parameter_str}) on qubit {qubit_str} was"
+                    " provided in the gate calibration set."
+                )
+
+        return gate_pulse_sequence(
+            **{p.name: v for p, v in zip(gate_pulse_sequence.parameters, parameters)}
+        )
+
+    def _find_parametric_gate_calibration(
+        self, gate: Gate, qubitset: QubitSet, number_assigned_values: int
+    ) -> PulseSequence | None:
+        for key in self._gate_calibrations.pulse_sequences:
+            if (
+                key[0].name == gate.name
+                and key[1] == qubitset
+                and sum(isinstance(param, FreeParameter) for param in key[0].parameters)
+                == number_assigned_values
+            ):
+                return self._gate_calibrations.pulse_sequences[key]
+
+    def _readout_frame(self, qubit: QubitSet) -> Frame:
+        readout_frame_names = {
+            "Rigetti": f"q{int(qubit)}_ro_rx_frame",
+            "Oxford": f"r{int(qubit)}_measure",
+        }
+        frame_name = readout_frame_names[self._device.provider_name]
+        return self._device.frames[frame_name]
+
+    @staticmethod
+    def _categorize_result_types(
+        result_types: list[ResultType],
+    ) -> list[ResultType]:
+        """
+        Categorize result types into result types with target and those without.
+
+        Args:
+            result_types (list[ResultType]): list of result types
+
+        Returns:
+            list[ResultType]: a list of result types with `target` attribute
+        """
+        target_result_types = []
+        for result_type in result_types:
+            if hasattr(result_type, "target"):
+                target_result_types.append(result_type)
+            else:
+                warnings.warn(
+                    f"{result_type} does not have have a pulse representation" " and it is ignored."
+                )
+        return target_result_types
+
+
+def _validate_device(device: AwsDevice | None) -> None:
+    if device is None:
+        raise ValueError("Device must be set before building pulse sequences.")
+    elif device.provider_name not in ("Rigetti", "Oxford"):
+        raise ValueError(f"Device {device.name} is not supported.")

src/braket/circuits/observable.py

@@ -27,6 +27,7 @@
     OpenQASMSerializationProperties,
     SerializationProperties,
 )
+from braket.pulse.pulse_sequence import PulseSequence
 from braket.registers.qubit_set import QubitSet
 
 
@@ -105,6 +106,17 @@ def _to_openqasm(
         """
         raise NotImplementedError("to_openqasm has not been implemented yet.")
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        """Returns the pulse sequence of the result type.
+
+        Args:
+            target (QubitSet | None): target qubit(s). Defaults to None.
+
+        Returns:
+            PulseSequence: A PulseSequence of the basis rotation for the corresponding observable.
+        """
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     @property
     def coefficient(self) -> int:
         """The coefficient of the observable.

src/braket/circuits/observables.py

@@ -30,6 +30,7 @@
     verify_quantum_operator_matrix_dimensions,
 )
 from braket.circuits.serialization import IRType, OpenQASMSerializationProperties
+from braket.pulse.pulse_sequence import PulseSequence
 from braket.registers.qubit_set import QubitSet
 
 
@@ -65,6 +66,9 @@ def to_matrix(self) -> np.ndarray:
             1.0 / np.sqrt(2.0) * np.array([[1.0, 1.0], [1.0, -1.0]], dtype=complex)
         )
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     @property
     def basis_rotation_gates(self) -> tuple[Gate, ...]:
         return tuple([Gate.Ry(-math.pi / 4)])  # noqa: C409
@@ -103,6 +107,9 @@ def _to_openqasm(
     def to_matrix(self) -> np.ndarray:
         return self.coefficient * np.eye(2, dtype=complex)
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     @property
     def basis_rotation_gates(self) -> tuple[Gate, ...]:
         return ()
@@ -150,6 +157,9 @@ def _to_openqasm(
         else:
             return f"{coef_prefix}x all"
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     def to_matrix(self) -> np.ndarray:
         return self.coefficient * np.array([[0.0, 1.0], [1.0, 0.0]], dtype=complex)
 
@@ -191,6 +201,9 @@ def _to_openqasm(
     def to_matrix(self) -> np.ndarray:
         return self.coefficient * np.array([[0.0, -1.0j], [1.0j, 0.0]], dtype=complex)
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     @property
     def basis_rotation_gates(self) -> tuple[Gate, ...]:
         return tuple([Gate.Z(), Gate.S(), Gate.H()])  # noqa: C409
@@ -229,6 +242,9 @@ def _to_openqasm(
     def to_matrix(self) -> np.ndarray:
         return self.coefficient * np.array([[1.0, 0.0], [0.0, -1.0]], dtype=complex)
 
+    def _to_pulse_sequence(self, target: QubitSet | None = None) -> PulseSequence:
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     @property
     def basis_rotation_gates(self) -> tuple[Gate, ...]:
         return ()

src/braket/circuits/result_type.py

@@ -23,6 +23,7 @@
     OpenQASMSerializationProperties,
     SerializationProperties,
 )
+from braket.pulse.pulse_sequence import PulseSequence
 from braket.registers.qubit import QubitInput
 from braket.registers.qubit_set import QubitSet, QubitSetInput
 
@@ -117,6 +118,14 @@ def _to_openqasm(self, serialization_properties: OpenQASMSerializationProperties
         """
         raise NotImplementedError("to_openqasm has not been implemented yet.")
 
+    def _to_pulse_sequence(self) -> PulseSequence:
+        """Returns the pulse sequence of the result type.
+
+        Returns:
+            PulseSequence: A PulseSequence corresponding to the result type.
+        """
+        raise NotImplementedError("_to_pulse_sequence has not been implemented yet.")
+
     def copy(
         self,
         target_mapping: dict[QubitInput, QubitInput] | None = None,

src/braket/circuits/result_types.py

@@ -28,6 +28,7 @@
     ResultType,
 )
 from braket.circuits.serialization import IRType, OpenQASMSerializationProperties
+from braket.pulse.pulse_sequence import PulseSequence
 from braket.registers.qubit_set import QubitSet, QubitSetInput
 
 """
@@ -492,6 +493,9 @@ def _to_openqasm(self, serialization_properties: OpenQASMSerializationProperties
         )
         return f"#pragma braket result expectation {observable_ir}"
 
+    def _to_pulse_sequence(self) -> PulseSequence:
+        return self.observable._to_pulse_sequence(target=self.target)
+
     @staticmethod
     @circuit.subroutine(register=True)
     def expectation(observable: Observable, target: QubitSetInput | None = None) -> ResultType:

src/braket/pulse/pulse_sequence.py

@@ -426,6 +426,37 @@ def __call__(
             param_values[str(key)] = val
         return self.make_bound_pulse_sequence(param_values)
 
+    def __iadd__(self, other: PulseSequence) -> PulseSequence:
+        """In-place concatenation of pulse sequence."""
+        if self._capture_v0_count:
+            raise ValueError(
+                "Cannot add a pulse sequence to another one that has already measurement"
+                " instructions."
+            )
+        for frameId, frame in other._frames.items():
+            _validate_uniqueness(self._frames, frame)
+            self._frames[frameId] = frame
+        for waveformId, waveform in other._waveforms.items():
+            _validate_uniqueness(self._waveforms, waveform)
+            self._waveforms[waveformId] = waveform
+        for fp in other._free_parameters:
+            if fp in self._free_parameters:
+                raise ValueError(
+                    f"A free parameter with the name {fp.name} already exists in the pulse"
+                    " sequence. Please rename this free parameter."
+                )
+            self._free_parameters.add(fp)
+
+        self._program += other._program
+        return self
+
+    def __add__(self, other: PulseSequence) -> PulseSequence:
+        """Return concatenation of two programs."""
+        assert isinstance(other, PulseSequence)
+        self_copy = deepcopy(self)
+        self_copy += other
+        return self_copy
+
     def __eq__(self, other: PulseSequence):
         sort_input_parameters = True
         return (