use std::collections::{HashMap, HashSet};
use std::ops::Range;
use std::str::FromStr;
use indexmap::IndexMap;
use numpy::{Complex64, PyArray2, ToPyArray};
use pyo3::{
exceptions::{PyDeprecationWarning, PyTypeError, PyUnicodeDecodeError, PyValueError},
prelude::*,
types::{PyBytes, PyFunction, PyRange, PyTuple},
PyErr,
};
use rigetti_pyo3::{create_init_submodule, impl_repr};
#[cfg(feature = "stubs")]
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum, gen_stub_pymethods};
use crate::{
instruction::{
quilpy::OwnedGateSignature, CalibrationDefinition, Declaration, DefaultHandler,
ExternPragmaMap, FrameAttributes, FrameIdentifier, Gate, Instruction,
MeasureCalibrationDefinition, Measurement, MemoryReference, Qubit, QubitPlaceholder,
Target, TargetPlaceholder, Waveform,
},
quil::Quil,
quilpy::{errors, impl_to_quil},
};
use super::{
analysis::{
BasicBlock, BasicBlockOwned, BasicBlockScheduleError, BasicBlockTerminator,
ControlFlowGraph, ControlFlowGraphOwned, QubitGraph, QubitGraphError,
},
scheduling::{ComputedScheduleItem, Schedule, Seconds, TimeSpan},
CalibrationExpansion, CalibrationSource, Calibrations, DefGateSequenceExpansion,
ExpansionResult, FrameSet, InstructionIndex, MemoryRegion, Program, Result, SourceMap,
SourceMapEntry, SourceMapIndexable,
};
create_init_submodule! {
classes: [
BasicBlockOwned, CalibrationExpansion,
CalibrationSource,
Calibrations, ControlFlowGraphOwned, FlatExpansionResult,
FrameSet,
InstructionSourceMap,
InstructionSourceMapEntry,
MemoryRegion,
OwnedDefGateSequenceExpansion,
Program,
PyScheduleSeconds,
ScheduleSecondsItem,
TimeSpanSeconds
],
complex_enums: [ CalibrationSource, FlatExpansionResult ],
errors: [
errors::ProgramError,
errors::ComputedScheduleError,
errors::BasicBlockScheduleError,
errors::QubitGraphError
],
}
impl_repr!(BasicBlockOwned);
impl_repr!(CalibrationExpansion);
impl_repr!(Calibrations);
impl_repr!(CalibrationSource);
impl_repr!(ControlFlowGraphOwned);
impl_repr!(FrameSet);
impl_repr!(FlatExpansionResult);
impl_repr!(MemoryRegion);
impl_repr!(Program);
impl_repr!(PyScheduleSeconds);
impl_repr!(ScheduleSecondsItem);
impl_repr!(TimeSpanSeconds);
impl_to_quil!(Program);
type ExpandedProgram = (Program, InstructionSourceMap);
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl Program {
#[staticmethod]
fn parse(input: &str) -> Result<Self> {
<Self as std::str::FromStr>::from_str(input)
}
#[gen_stub(override_return_type(
type_repr = "builtins.dict[typing.Optional[builtins.str], builtins.list[_quil.instructions.Instruction]]",
imports = ("quil._quil", "builtins", "typing"),
))]
#[getter]
fn pragma_extern_map(&self) -> ExternPragmaMap {
self.extern_pragma_map.clone()
}
#[gen_stub(override_return_type(
type_repr = "builtins.set[_quil.instructions.Qubit]",
imports = ("quil._quil", "builtins"),
))]
#[getter]
fn used_qubits(&self) -> HashSet<Qubit> {
self.used_qubits.clone()
}
#[gen_stub(override_return_type(
type_repr = "builtins.dict[builtins.str, _quil.instructions.Waveform]",
imports = ("quil._quil", "builtins"),
))]
#[getter]
fn waveforms(&self) -> IndexMap<String, Waveform> {
self.waveforms.clone()
}
#[setter]
fn set_waveforms(
&mut self,
#[gen_stub(override_type(
type_repr = "typing.Mapping[builtins.str, _quil.instructions.Waveform]",
imports = ("quil._quil", "typing"),
))]
waveforms: IndexMap<String, Waveform>,
) {
self.waveforms = waveforms;
}
#[getter(body_instructions)]
fn py_body_instructions(&self) -> Vec<Instruction> {
self.instructions.clone()
}
fn copy(&self) -> Self {
self.clone()
}
fn control_flow_graph(&self) -> ControlFlowGraphOwned {
ControlFlowGraphOwned::from(ControlFlowGraph::from(self))
}
#[getter]
fn declarations(&self) -> HashMap<String, Declaration> {
self.to_instructions()
.into_iter()
.filter_map(|inst| match inst {
Instruction::Declaration(declaration) => {
Some((declaration.name.clone(), declaration))
}
_ => None,
})
.collect()
}
#[pyo3(name = "expand_calibrations_with_source_map")]
fn py_expand_calibrations_with_source_map(&self) -> Result<ExpandedProgram> {
let (expanded, source_map) = self.expand_calibrations_with_source_map()?;
Ok((expanded, source_map.into()))
}
#[pyo3(
signature = (predicate = None),
name = "expand_defgate_sequences_with_source_map",
)]
fn py_expand_defgate_sequences_with_source_map<'py>(
&self,
py: Python<'py>,
#[gen_stub(
override_type(
type_repr="typing.Optional[collections.abc.Callable[[builtins.str], builtins.bool]]",
imports=("collections.abc", "builtins", "typing")
)
)]
predicate: Option<&Bound<'py, PyFunction>>,
) -> Result<ExpandedProgram> {
let (expanded_program, source_map) =
self.expand_defgate_sequences_with_source_map(|key: &str| -> bool {
predicate.is_none_or(|f| match call_user_func(py, f, key) {
Ok(val) => val,
Err(err) => panic!("error calling predicate: {err}"),
})
})?;
Ok((expanded_program, source_map.into()))
}
#[pyo3(signature = (predicate = None), name = "expand_defgate_sequences")]
fn py_expand_defgate_sequences<'py>(
&self,
py: Python<'py>,
#[gen_stub(
override_type(
type_repr="typing.Optional[collections.abc.Callable[[builtins.str], builtins.bool]]",
imports=("collections.abc", "builtins", "typing")
)
)]
predicate: Option<&Bound<'py, PyAny>>,
) -> Result<Self> {
self.clone().expand_defgate_sequences(|key: &str| -> bool {
predicate.is_none_or(|f| match call_user_func(py, f, key) {
Ok(val) => val,
Err(err) => panic!("error calling predicate: {err}"),
})
})
}
#[pyo3(name = "add_instructions")]
fn py_add_instructions(
&mut self,
#[gen_stub(override_type(
type_repr = "typing.Sequence[_quil.instructions.Instruction]",
imports = ("quil._quil", "typing"),
))]
instructions: Vec<Instruction>,
) {
self.add_instructions(instructions);
}
#[pyo3(name = "filter_instructions")]
fn py_filter_instructions<'py>(
&self,
py: Python<'py>,
#[gen_stub(
override_type(
type_repr="collections.abc.Callable[[_quil.instructions.Instruction], builtins.bool]",
imports=("quil._quil", "collections.abc", "builtins")
)
)]
predicate: &Bound<'py, PyAny>,
) -> Self {
self.filter_instructions(|inst| match call_user_func(py, predicate, inst.clone()) {
Ok(val) => val,
Err(err) => panic!("error calling predicate: {err}"),
})
}
#[pyo3(
name = "resolve_placeholders_with_custom_resolvers",
signature = (*, target_resolver = None, qubit_resolver = None)
)]
fn py_resolve_placeholders_with_custom_resolvers(
&mut self,
#[gen_stub(override_type(
type_repr = "typing.Optional[collections.abc.Callable[[_quil.instructions.TargetPlaceholder], typing.Optional[builtins.str]]]",
imports = ("quil._quil", "collections.abc", "builtins", "typing")
))]
target_resolver: Option<PyTargetResolver>,
#[gen_stub(override_type(
type_repr = "typing.Optional[collections.abc.Callable[[_quil.instructions.QubitPlaceholder], typing.Optional[builtins.int]]]",
imports = ("quil._quil", "collections.abc", "builtins", "typing")
))]
qubit_resolver: Option<PyQubitResolver>,
) {
#[allow(clippy::type_complexity)]
let rs_qubit_resolver: Box<dyn Fn(&QubitPlaceholder) -> Option<u64>> =
if let Some(resolver) = qubit_resolver {
Box::new(move |placeholder: &QubitPlaceholder| -> Option<u64> {
Python::attach(|py| {
let placeholder = placeholder
.clone()
.into_pyobject(py)
.expect("QubitPlaceholder.into_python() should be infallible");
let resolved_qubit =
resolver.0.call1(py, (placeholder,)).unwrap_or_else(|err| {
panic!("qubit_resolver returned an error: {err}")
});
resolved_qubit.extract(py).unwrap_or_else(|err| {
panic!("qubit_resolver must return None or int: {err}")
})
})
})
} else {
self.default_qubit_resolver()
};
#[allow(clippy::type_complexity)]
let rs_target_resolver: Box<dyn Fn(&TargetPlaceholder) -> Option<String>> =
if let Some(resolver) = target_resolver {
Box::new(move |placeholder: &TargetPlaceholder| -> Option<String> {
Python::attach(|py| {
let placeholder = placeholder
.clone()
.into_pyobject(py)
.expect("TargetPlaceholder.into_python() should be infallible");
let resolved_target =
resolver.0.call1(py, (placeholder,)).unwrap_or_else(|err| {
panic!("target_resolver returned an error: {err}")
});
resolved_target.extract(py).unwrap_or_else(|err| {
panic!("target_resolver must return None or str: {err}")
})
})
})
} else {
self.default_target_resolver()
};
self.resolve_placeholders_with_custom_resolvers(rs_target_resolver, rs_qubit_resolver);
}
#[pyo3(name = "into_simplified")]
fn py_into_simplified(&self) -> Result<Self> {
self.simplify(&DefaultHandler)
}
#[pyo3(name = "to_unitary")]
fn py_to_unitary<'py>(
&self,
py: Python<'py>,
n_qubits: u64,
) -> PyResult<Bound<'py, PyArray2<Complex64>>> {
Ok(self.to_unitary(n_qubits)?.to_pyarray(py))
}
#[gen_stub(skip)]
#[pyo3(
name = "wrap_in_loop",
signature = (loop_count_reference, start_target, end_target = None, iterations = None),
)]
fn backwards_compatible_py_wrap_in_loop(
&self,
py: pyo3::Python<'_>,
loop_count_reference: MemoryReference,
start_target: Target,
end_target: Option<EndTargetParam>,
iterations: Option<u32>,
) -> pyo3::PyResult<Self> {
let iterations = match (end_target, iterations) {
(Some(EndTargetParam::Iterations(iterations)), None) | (None, Some(iterations)) => {
iterations
}
(Some(EndTargetParam::Target(_)), Some(iterations)) => {
PyErr::warn(
py,
&py.get_type::<PyDeprecationWarning>(),
c"`end_target` is deprecated and will be ignored",
1,
)?;
iterations
}
(Some(EndTargetParam::Target(_)) | None, None) => {
return Err(PyTypeError::new_err(
"wrap_in_loop missing required positional argument: 'iterations'",
));
}
(Some(EndTargetParam::Iterations(_)), Some(_)) => {
return Err(PyTypeError::new_err(
"wrap_in_loop got multiple values for argument: 'iterations'",
));
}
};
if iterations == 0 {
return Err(PyValueError::new_err(
"wrap_in_loop got a value of 0 for 'iterations'; \
this will produce an empty Program in future releases",
));
}
Ok(self.wrap_in_loop(loop_count_reference, start_target, iterations))
}
fn __add__(&self, rhs: Self) -> Self {
self.clone() + rhs
}
fn __iadd__(&mut self, rhs: Self) {
*self += rhs;
}
fn __getstate__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
Ok(PyBytes::new(py, self.to_quil()?.as_bytes()))
}
pub fn __setstate__(&mut self, py: Python<'_>, state: &Bound<'_, PyBytes>) -> PyResult<()> {
let quil = std::str::from_utf8(state.as_bytes())
.map_err(|e| PyUnicodeDecodeError::new_err_from_utf8(py, state.as_bytes(), e))?;
*self = Self::from_str(quil)?;
Ok(())
}
}
#[derive(pyo3::FromPyObject)]
enum EndTargetParam {
#[expect(
dead_code,
reason = "the field isn't read, but we still want it type-checked"
)]
#[pyo3(annotation = "quil.instructions.Target")]
Target(Target),
#[pyo3(annotation = "int")]
Iterations(u32),
}
#[cfg(feature = "stubs")]
pyo3_stub_gen::inventory::submit! {
pyo3_stub_gen::derive::gen_methods_from_python! {
r#"
import builtins
from quil._quil import instructions
class Program:
@overload
def wrap_in_loop(
self,
loop_count_reference: _quil.instructions.MemoryReference,
start_target: _quil.instructions.Target,
iterations: builtins.int,
) -> Program:
r"""
Return a copy of the [`Program`] wrapped in a loop that repeats `iterations` times.
The loop is constructed by wrapping the body of the program in classical Quil instructions.
The given `loop_count_reference` must refer to an INTEGER memory region. The value at the
reference given will be set to `iterations` and decremented in the loop. The loop will
terminate when the reference reaches 0. For this reason your program should not itself
modify the value at the reference unless you intend to modify the remaining number of
iterations (i.e. to break the loop).
The given `start_target` will be used as the entry point for the loop. You should provide
[`Target::Placeholder`] or otherwise choose a unique [`Target`] that won't be used elsewhere
in the program.
If `iterations` is `1`, then a copy of the program is returned without any changes.
**Note:** For backwards compatibility with earlier releases of `quil-rs`, this function has
three behaviors that will change in future releases.
1. If `iterations` is 0, then an error is raised.
- In the future, an empty program will be returned – specifically, [a program with no
body instructions, but retaining all of the calibrations,
etc.][Self::clone_without_body_instructions].
- In prior releases of `quil-rs`, the input program was returned unmodified instead, as
though `iterations` had been `1`.
2. This function additionally accepts an `end_target` argument either by name or
positionally before `iterations`. This argument is no longer used, and its presence will
raise a `DeprecationWarning`. In future releases, providing this argument will be an
error.
- As an edge case, it is currently possible to call this function as
`program.wrap_in_loop(loop_count_reference, start_target, end_target = 3)` and get a
3-iteration loop. This behavior is not supported and will be removed without warning
in a future release of Quil.
3. Either or both of the `end_target` and `iterations` parameters can be given as `None`
instead of being omitted. This is the same as being omitted, which is or is not allowed
as provided above. In future releases, `None` will not be accepted.
r"""
@overload
@deprecated("the `end_target` parameter is deprecated and will be ignored")
def wrap_in_loop(
self,
loop_count_reference: _quil.instructions.MemoryReference,
start_target: _quil.instructions.Target,
end_target: _quil.instructions.Target,
iterations: int,
) -> Program: ...
@overload
@deprecated("the `iterations` parameter must be provided")
def wrap_in_loop(
self,
loop_count_reference: _quil.instructions.MemoryReference,
start_target: _quil.instructions.Target,
end_target: None,
iterations: None,
) -> Program: ...
@overload
@deprecated("the `iterations` parameter must be provided")
def wrap_in_loop(
self,
loop_count_reference: _quil.instructions.MemoryReference,
start_target: _quil.instructions.Target,
) -> Program: ...
"#
}
}
#[derive(Debug, thiserror::Error)]
enum UserFunctionError {
#[error("calling the function resulted in an error: {0}")]
CallError(#[source] PyErr),
#[error("function returned the wrong type")]
TypeError(#[source] PyErr),
}
#[inline]
fn call_user_func<'a, 'py, T, U>(
py: Python<'py>,
user_func: &'a Bound<'py, PyAny>,
param: T,
) -> std::result::Result<U, UserFunctionError>
where
T: IntoPyObject<'py>,
U: FromPyObjectOwned<'py>,
{
let args: Bound<'py, PyTuple> = (param,)
.into_pyobject(py)
.expect("Tuple::into_pyobject() should be infallible");
user_func
.call1(args)
.map_err(UserFunctionError::CallError)?
.extract::<U>()
.map_err(Into::<PyErr>::into)
.map_err(UserFunctionError::TypeError)
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl Calibrations {
#[new]
fn new(
calibrations: Vec<CalibrationDefinition>,
measure_calibrations: Vec<MeasureCalibrationDefinition>,
) -> Self {
Self {
calibrations: calibrations.into(),
measure_calibrations: measure_calibrations.into(),
}
}
#[getter(calibrations)]
fn py_calibrations(&self) -> Vec<CalibrationDefinition> {
self.iter_calibrations().cloned().collect()
}
#[getter(measure_calibrations)]
fn py_measure_calibrations(&self) -> Vec<MeasureCalibrationDefinition> {
self.iter_measure_calibrations().cloned().collect()
}
#[pyo3(name = "expand")]
fn py_expand(
&self,
instruction: &Instruction,
previous_calibrations: Vec<Instruction>,
) -> Result<Option<Vec<Instruction>>> {
self.expand(instruction, &previous_calibrations)
}
#[pyo3(name = "get_match_for_measurement")]
fn py_get_match_for_measurement(
&self,
measurement: &Measurement,
) -> Option<MeasureCalibrationDefinition> {
self.get_match_for_measurement(measurement).cloned()
}
#[pyo3(name = "get_match_for_gate")]
fn py_get_match_for_gate(&self, gate: &Gate) -> Option<CalibrationDefinition> {
self.get_match_for_gate(gate).cloned()
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[pymethods]
impl CalibrationExpansion {
#[gen_stub(override_return_type(type_repr = "range"))]
#[getter(range)]
pub fn py_range<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyRange>> {
PyRange::new(
py,
self.range.start.0.try_into()?,
self.range.end.0.try_into()?,
)
}
#[getter(expansions)]
fn py_expansions(&self) -> InstructionSourceMap {
(&self.expansions).into()
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl FrameSet {
#[pyo3(name = "get")]
fn py_get(&self, identifier: &FrameIdentifier) -> Option<FrameAttributes> {
self.get(identifier).cloned()
}
#[pyo3(name = "get_keys")]
fn py_get_keys(&self) -> Vec<FrameIdentifier> {
self.get_keys().into_iter().cloned().collect()
}
fn get_all_frames(&self) -> HashMap<FrameIdentifier, FrameAttributes> {
self.frames.clone()
}
#[pyo3(name = "intersection")]
pub fn py_intersection(&self, identifiers: HashSet<FrameIdentifier>) -> Self {
self.intersection(&identifiers)
}
}
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl CalibrationSource {
#[gen_stub(override_return_type(
type_repr = "builtins.tuple[_quil.instructions.CalibrationIdentifier | _quil.instructions.MeasureCalibrationIdentifier]",
imports = ("quil._quil.instructions", "builtins")
))]
fn __getnewargs__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyTuple>> {
match self {
Self::Calibration(value) => (value.clone(),).into_pyobject(py),
Self::MeasureCalibration(value) => (value.clone(),).into_pyobject(py),
}
}
}
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyclass(
name = "InstructionSourceMap",
module = "quil._quil.program",
frozen,
from_py_object
)]
pub(crate) struct InstructionSourceMap(SourceMap<InstructionIndex, FlatExpansionResult>);
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyclass(
name = "InstructionSourceMapEntry",
module = "quil._quil.program",
frozen,
from_py_object
)]
pub(crate) struct InstructionSourceMapEntry(SourceMapEntry<InstructionIndex, FlatExpansionResult>);
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl InstructionSourceMapEntry {
pub fn source_location(&self) -> InstructionIndex {
*self.0.source_location()
}
pub fn target_location(&self) -> FlatExpansionResult {
self.0.target_location().clone()
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl InstructionSourceMap {
fn entries(&self) -> Vec<InstructionSourceMapEntry> {
self.0
.entries
.iter()
.cloned()
.map(InstructionSourceMapEntry)
.collect()
}
fn list_sources_for_target_index(
&self,
target_index: InstructionIndex,
) -> Vec<&InstructionIndex> {
self.0.list_sources(&target_index)
}
fn list_sources_for_calibration_used(
&self,
calibration_used: CalibrationSource,
) -> Vec<&InstructionIndex> {
self.0.list_sources(&calibration_used)
}
pub fn list_sources_for_gate_expansion(
&self,
gate_signature: OwnedGateSignature,
) -> Vec<&InstructionIndex> {
self.0.list_sources(&gate_signature)
}
fn list_targets_for_source_index(
&self,
source_index: InstructionIndex,
) -> Vec<FlatExpansionResult> {
self.0
.list_targets(&source_index)
.into_iter()
.cloned()
.collect()
}
}
impl<R: Into<FlatExpansionResult> + Clone> From<SourceMap<InstructionIndex, ExpansionResult<R>>>
for InstructionSourceMap
{
fn from(value: SourceMap<InstructionIndex, ExpansionResult<R>>) -> Self {
let entries = value
.entries
.into_iter()
.map(|entry| SourceMapEntry {
source_location: entry.source_location,
target_location: FlatExpansionResult::from(entry.target_location),
})
.collect();
InstructionSourceMap(SourceMap::new(entries))
}
}
impl<R: Into<FlatExpansionResult> + Clone> From<&SourceMap<InstructionIndex, ExpansionResult<R>>>
for InstructionSourceMap
{
fn from(value: &SourceMap<InstructionIndex, ExpansionResult<R>>) -> Self {
let entries = value
.entries
.iter()
.map(|entry| SourceMapEntry {
source_location: *entry.source_location(),
target_location: FlatExpansionResult::from(entry.target_location()),
})
.collect();
InstructionSourceMap(SourceMap::new(entries))
}
}
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyo3::pyclass(module = "quil._quil.program", eq, from_py_object)]
pub struct OwnedDefGateSequenceExpansion {
#[pyo3(get)]
source_signature: OwnedGateSignature,
range: Range<InstructionIndex>,
nested_expansions: SourceMap<InstructionIndex, ExpansionResult<OwnedDefGateSequenceExpansion>>,
}
impl<'a> From<&'a DefGateSequenceExpansion<'a>> for OwnedDefGateSequenceExpansion {
fn from(value: &'a DefGateSequenceExpansion<'a>) -> Self {
Self {
source_signature: value.source_signature().clone().into(),
range: value.range().clone(),
nested_expansions: SourceMap::new(
value
.nested_expansions()
.entries()
.iter()
.map(|entry| {
let target_location = entry.target_location();
SourceMapEntry {
source_location: *entry.source_location(),
target_location: match target_location {
ExpansionResult::Unmodified(index) => {
ExpansionResult::Unmodified(*index)
}
ExpansionResult::Rewritten(rewrite) => ExpansionResult::Rewritten(
OwnedDefGateSequenceExpansion::from(rewrite),
),
},
}
})
.collect(),
),
}
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[pymethods]
impl OwnedDefGateSequenceExpansion {
#[gen_stub(override_return_type(type_repr = "range"))]
#[getter(range)]
pub fn py_range<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyRange>> {
PyRange::new(
py,
self.range.start.0.try_into()?,
self.range.end.0.try_into()?,
)
}
#[pyo3(name = "expansions")]
fn py_expansions(&self) -> InstructionSourceMap {
(&self.nested_expansions).into()
}
}
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass_complex_enum)]
#[pyo3::pyclass(
name = "InstructionTarget",
module = "quil._quil.program",
eq,
frozen,
from_py_object
)]
pub enum FlatExpansionResult {
Unmodified(InstructionIndex),
Calibration(CalibrationExpansion),
DefGateSequence(OwnedDefGateSequenceExpansion),
}
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl FlatExpansionResult {
#[gen_stub(override_return_type(type_repr = "tuple[CalibrationExpansion, builtins.int]"))]
fn __getnewargs__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyTuple>> {
match self {
Self::Unmodified(value) => (value,).into_pyobject(py),
Self::Calibration(value) => (value.clone(),).into_pyobject(py),
Self::DefGateSequence(value) => (value.clone(),).into_pyobject(py),
}
}
}
impl From<CalibrationExpansion> for FlatExpansionResult {
fn from(value: CalibrationExpansion) -> Self {
Self::Calibration(value)
}
}
impl<'a> From<DefGateSequenceExpansion<'a>> for FlatExpansionResult {
fn from(value: DefGateSequenceExpansion<'a>) -> Self {
Self::DefGateSequence(OwnedDefGateSequenceExpansion::from(&value))
}
}
impl From<OwnedDefGateSequenceExpansion> for FlatExpansionResult {
fn from(value: OwnedDefGateSequenceExpansion) -> Self {
Self::DefGateSequence(value)
}
}
impl<R: Into<FlatExpansionResult> + Clone> From<&ExpansionResult<R>> for FlatExpansionResult {
fn from(value: &ExpansionResult<R>) -> Self {
match value {
ExpansionResult::Unmodified(index) => Self::Unmodified(*index),
ExpansionResult::Rewritten(rewrite) => rewrite.clone().into(),
}
}
}
impl<R: Into<FlatExpansionResult>> From<ExpansionResult<R>> for FlatExpansionResult {
fn from(value: ExpansionResult<R>) -> Self {
match value {
ExpansionResult::Unmodified(index) => Self::Unmodified(index),
ExpansionResult::Rewritten(rewrite) => rewrite.into(),
}
}
}
impl SourceMapIndexable<InstructionIndex> for FlatExpansionResult {
fn contains(&self, other: &InstructionIndex) -> bool {
match self {
Self::Unmodified(index) => index == other,
Self::Calibration(expansion) => expansion.contains(other),
Self::DefGateSequence(expansion) => expansion.contains(other),
}
}
}
impl SourceMapIndexable<CalibrationSource> for FlatExpansionResult {
fn contains(&self, other: &CalibrationSource) -> bool {
if let Self::Calibration(expansion) = self {
expansion.contains(other)
} else {
false
}
}
}
impl SourceMapIndexable<InstructionIndex> for OwnedDefGateSequenceExpansion {
fn contains(&self, other: &InstructionIndex) -> bool {
self.range.contains(other)
}
}
impl SourceMapIndexable<OwnedGateSignature> for OwnedDefGateSequenceExpansion {
fn contains(&self, other: &OwnedGateSignature) -> bool {
&self.source_signature == other
}
}
impl<R> SourceMapIndexable<OwnedGateSignature> for ExpansionResult<R>
where
R: SourceMapIndexable<OwnedGateSignature>,
{
fn contains(&self, other: &OwnedGateSignature) -> bool {
if let Self::Rewritten(rewrite) = self {
rewrite.contains(other)
} else {
false
}
}
}
impl SourceMapIndexable<OwnedGateSignature> for FlatExpansionResult {
fn contains(&self, other: &OwnedGateSignature) -> bool {
if let Self::DefGateSequence(expansion) = self {
expansion.contains(other)
} else {
false
}
}
}
#[derive(FromPyObject, IntoPyObject, IntoPyObjectRef)]
struct PyQubitResolver(Py<PyFunction>);
#[derive(FromPyObject, IntoPyObject, IntoPyObjectRef)]
struct PyTargetResolver(Py<PyFunction>);
#[cfg(feature = "stubs")]
mod stubs {
use pyo3_stub_gen::{PyStubType, TypeInfo};
use super::{InstructionIndex, Seconds};
impl PyStubType for InstructionIndex {
fn type_output() -> TypeInfo {
TypeInfo::builtin("int")
}
}
impl PyStubType for Seconds {
fn type_output() -> TypeInfo {
TypeInfo::builtin("float")
}
}
}
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyclass(
name = "ScheduleSeconds",
module = "quil._quil.program",
subclass,
eq,
frozen,
from_py_object
)]
pub(crate) struct PyScheduleSeconds(pub Schedule<Seconds>);
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyclass(
name = "ScheduleSecondsItem",
module = "quil._quil.program",
subclass,
eq,
frozen,
from_py_object
)]
pub(crate) struct ScheduleSecondsItem(pub ComputedScheduleItem<Seconds>);
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "stubs", gen_stub_pyclass)]
#[pyclass(
name = "TimeSpanSeconds",
module = "quil._quil.program",
subclass,
eq,
frozen,
from_py_object
)]
pub(crate) struct TimeSpanSeconds(pub TimeSpan<Seconds>);
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl PyScheduleSeconds {
#[getter]
fn items(&self) -> Vec<ScheduleSecondsItem> {
self.0
.items
.iter()
.map(|item| ScheduleSecondsItem(item.clone()))
.collect()
}
#[getter]
fn duration(&self) -> Seconds {
self.0.duration().clone()
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl ScheduleSecondsItem {
#[getter]
fn time_span(&self) -> TimeSpanSeconds {
TimeSpanSeconds(self.0.time_span.clone())
}
#[getter]
fn instruction_index(&self) -> usize {
self.0.instruction_index
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl TimeSpanSeconds {
#[getter]
fn start(&self) -> Seconds {
self.0.start_time().clone()
}
#[getter]
fn duration(&self) -> Seconds {
self.0.duration().clone()
}
#[getter]
fn end(&self) -> Seconds {
self.0.end()
}
}
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl ControlFlowGraphOwned {
#[new]
fn new(instance: Self) -> Self {
instance
}
fn has_dynamic_control_flow(&self) -> bool {
ControlFlowGraph::from(self).has_dynamic_control_flow()
}
fn basic_blocks(&self) -> Vec<BasicBlockOwned> {
self.blocks.clone()
}
}
#[cfg_attr(not(feature = "stubs"), optipy::strip_pyo3(only_stubs))]
#[cfg_attr(feature = "stubs", gen_stub_pymethods)]
#[pymethods]
impl BasicBlockOwned {
#[new]
fn new(instance: Self) -> Self {
instance
}
fn as_schedule_seconds(
&self,
program: &Program,
) -> std::result::Result<PyScheduleSeconds, BasicBlockScheduleError> {
BasicBlock::from(self)
.as_schedule_seconds(program, &DefaultHandler)
.map(PyScheduleSeconds)
}
#[expect(clippy::doc_markdown)]
fn gate_depth(
&self,
gate_minimum_qubit_count: usize,
) -> std::result::Result<usize, QubitGraphError> {
let block = BasicBlock::from(self);
QubitGraph::try_from_basic_block(&block, &DefaultHandler)
.map(|graph| graph.gate_depth(gate_minimum_qubit_count))
}
#[gen_stub(override_return_type(
type_repr = "typing.Optional[_quil.instructions.Target]",
imports = ("quil._quil", "typing"),
))]
#[getter]
fn label(&self) -> Option<Target> {
self.label.clone()
}
#[gen_stub(override_return_type(
type_repr = "typing.Optional[_quil.instructions.Instruction]",
imports = ("quil._quil", "typing"),
))]
#[getter]
fn terminator(&self) -> Option<Instruction> {
BasicBlockTerminator::from(&self.terminator).into_instruction()
}
}