use crate::operations::convert_operation_to_pyobject;
use crate::{convert_into_circuit, CircuitWrapper};
use ndarray::{Array1, Array2};
use num_complex::Complex64;
use numpy::{PyArray1, PyArray2, PyReadonlyArray1, PyReadonlyArray2, ToPyArray};
use pyo3::exceptions::{PyRuntimeError, PyTypeError};
use pyo3::prelude::*;
use pyo3::types::PyByteArray;
use pyo3::types::PySet;
use qoqo_calculator::CalculatorFloat;
use qoqo_calculator_pyo3::{convert_into_calculator_float, CalculatorFloatWrapper};
use qoqo_macros::*;
use roqoqo::operations::*;
use roqoqo::Circuit;
#[cfg(feature = "json_schema")]
use roqoqo::ROQOQO_VERSION;
use std::collections::HashMap;
#[wrap(Operate, OperatePragma, JsonSchema)]
#[derive(Eq)]
struct PragmaSetNumberOfMeasurements {
number_measurements: usize,
readout: String,
}
#[wrap(Operate, OperatePragma, JsonSchema)]
pub struct PragmaLoop {
repetitions: CalculatorFloat,
circuit: Circuit,
}
#[pymodule]
fn pragma_set_statevector(_py: Python, module: &Bound<PyModule>) -> PyResult<()> {
module.add_class::<PragmaSetStateVectorWrapper>()?;
Ok(())
}
#[pyclass(
from_py_object,
name = "PragmaSetStateVector",
module = "qoqo.operations"
)]
#[derive(Clone, Debug, PartialEq)]
pub struct PragmaSetStateVectorWrapper {
pub internal: PragmaSetStateVector,
}
insert_pyany_to_operation!(
"PragmaSetStateVector" =>{
let array = op.call_method0("statevector").map_err(|_| QoqoError::ConversionError)?;
let statevec_casted: PyReadonlyArray1<Complex64> = array.extract().unwrap();
let statevec_array: Array1<Complex64> = statevec_casted.as_array().to_owned();
Ok(PragmaSetStateVector::new(statevec_array).into())
}
);
insert_operation_to_pyobject!(
Operation::PragmaSetStateVector(internal) => {
{
let pyref: Py<PragmaSetStateVectorWrapper> =
Py::new(py, PragmaSetStateVectorWrapper { internal }).unwrap();
pyref.into_pyobject(py).map(|bound| bound.as_any().to_owned()).map_err(|_| PyValueError::new_err("Unable to convert to Python object"))
}
}
);
#[pymethods]
impl PragmaSetStateVectorWrapper {
#[new]
fn new(statevector: &Bound<PyAny>) -> PyResult<Self> {
let try_cast: PyResult<Array1<Complex64>> =
if let Ok(extracted) = statevector.extract::<PyReadonlyArray1<Complex64>>() {
let statevec: Array1<Complex64> = extracted.as_array().to_owned();
Ok(statevec)
} else if let Ok(extracted) = statevector.extract::<PyReadonlyArray1<f64>>() {
let statevec: Array1<f64> = extracted.as_array().to_owned();
let statevec: Array1<Complex64> = statevec
.into_iter()
.map(|f| Complex64::new(f, 0.0))
.collect();
Ok(statevec)
} else if let Ok(extracted) = statevector.extract::<PyReadonlyArray1<isize>>() {
let statevec: Array1<isize> = extracted.as_array().to_owned();
let statevec: Array1<Complex64> = statevec
.into_iter()
.map(|f| Complex64::new(f as f64, 0.0))
.collect();
Ok(statevec)
} else {
Err(PyTypeError::new_err(
"Internal error: no successful PyReadonlyArray1 extraction.",
))
};
match try_cast {
Ok(array) => Ok(Self {
internal: PragmaSetStateVector::new(array),
}),
Err(_) => {
let statevec_casted: Vec<Complex64> = Vec::extract(statevector.as_borrowed())?;
let statevec_array: Array1<Complex64> = Array1::from(statevec_casted);
Ok(Self {
internal: PragmaSetStateVector::new(statevec_array),
})
}
}
}
fn statevector(&self) -> Py<PyArray1<Complex64>> {
Python::attach(|py| -> Py<PyArray1<Complex64>> {
self.internal.statevector().to_pyarray(py).unbind()
})
}
fn involved_qubits<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PySet>> {
PySet::new(py, ["All"])?
.into_pyobject(py)
.map_err(|_| PyRuntimeError::new_err("Unable to convert to Python object"))
}
fn tags(&self) -> Vec<String> {
self.internal.tags().iter().map(|s| s.to_string()).collect()
}
fn hqslang(&self) -> &'static str {
self.internal.hqslang()
}
fn is_parametrized(&self) -> bool {
self.internal.is_parametrized()
}
fn substitute_parameters(
&self,
substitution_parameters: std::collections::HashMap<String, f64>,
) -> PyResult<Self> {
let mut calculator = qoqo_calculator::Calculator::new();
for (key, val) in substitution_parameters.iter() {
calculator.set_variable(key, *val);
}
Ok(Self {
internal: self
.internal
.substitute_parameters(&calculator)
.map_err(|x| {
pyo3::exceptions::PyRuntimeError::new_err(format!(
"Parameter Substitution failed: {x:?}"
))
})?,
})
}
fn remap_qubits(&self, mapping: std::collections::HashMap<usize, usize>) -> PyResult<Self> {
let new_internal = self
.internal
.remap_qubits(&mapping)
.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Qubit remapping failed: "))?;
Ok(Self {
internal: new_internal,
})
}
fn __copy__(&self) -> PragmaSetStateVectorWrapper {
self.clone()
}
fn __deepcopy__(&self, _memodict: &Bound<PyAny>) -> PragmaSetStateVectorWrapper {
self.clone()
}
fn __format__(&self, _format_spec: &str) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __repr__(&self) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __richcmp__(
&self,
other: &Bound<PyAny>,
op: pyo3::class::basic::CompareOp,
) -> PyResult<bool> {
let other = crate::operations::convert_pyany_to_operation(other).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Right hand side cannot be converted to Operation",
)
})?;
match op {
pyo3::class::basic::CompareOp::Eq => {
Ok(Operation::from(self.internal.clone()) == other)
}
pyo3::class::basic::CompareOp::Ne => {
Ok(Operation::from(self.internal.clone()) != other)
}
_ => Err(pyo3::exceptions::PyNotImplementedError::new_err(
"Other comparison not implemented.",
)),
}
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn json_schema() -> String {
let schema = schemars::schema_for!(PragmaSetStateVector);
serde_json::to_string_pretty(&schema).expect("Unexpected failure to serialize schema")
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn current_version() -> String {
ROQOQO_VERSION.to_string()
}
#[cfg(feature = "json_schema")]
pub fn min_supported_version(&self) -> String {
let min_version: (u32, u32, u32) =
PragmaSetStateVector::minimum_supported_roqoqo_version(&self.internal);
format!("{}.{}.{}", min_version.0, min_version.1, min_version.2)
}
}
#[pymodule]
fn pragma_set_density_matrix(_py: Python, module: &Bound<PyModule>) -> PyResult<()> {
module.add_class::<PragmaSetDensityMatrixWrapper>()?;
Ok(())
}
#[pyclass(
from_py_object,
name = "PragmaSetDensityMatrix",
module = "qoqo.operations"
)]
#[derive(Clone, Debug, PartialEq)]
pub struct PragmaSetDensityMatrixWrapper {
pub internal: PragmaSetDensityMatrix,
}
insert_pyany_to_operation!(
"PragmaSetDensityMatrix" =>{
let array = op.call_method0("density_matrix").map_err(|_| QoqoError::ConversionError)?;
let density_matrix_op: PyReadonlyArray2<Complex64> = array.extract().unwrap();
let density_matrix: Array2<Complex64> = density_matrix_op.as_array().to_owned();
Ok(PragmaSetDensityMatrix::new(density_matrix).into())
}
);
insert_operation_to_pyobject!(
Operation::PragmaSetDensityMatrix(internal) => {
{
let pyref: Py<PragmaSetDensityMatrixWrapper> =
Py::new(py, PragmaSetDensityMatrixWrapper { internal }).unwrap();
pyref.into_pyobject(py).map(|bound| bound.as_any().to_owned()).map_err(|_| PyValueError::new_err("Unable to convert to Python object"))
}
}
);
#[pymethods]
impl PragmaSetDensityMatrixWrapper {
#[new]
fn new(density_matrix: &Bound<PyAny>) -> PyResult<Self> {
let try_cast: PyResult<Array2<Complex64>> =
if let Ok(extracted) = density_matrix.extract::<PyReadonlyArray2<Complex64>>() {
let matrix: Array2<Complex64> = extracted.as_array().to_owned();
Ok(matrix)
} else if let Ok(extracted) = density_matrix.extract::<PyReadonlyArray2<f64>>() {
let matrix: Array2<f64> = extracted.as_array().to_owned();
let matrix: Array2<Complex64> = matrix.map(|f| Complex64::new(*f, 0.0));
Ok(matrix)
} else if let Ok(extracted) = density_matrix.extract::<PyReadonlyArray2<isize>>() {
let matrix: Array2<isize> = extracted.as_array().to_owned();
let matrix: Array2<Complex64> = matrix.map(|f| Complex64::new((*f) as f64, 0.0));
Ok(matrix)
} else {
Err(PyTypeError::new_err(
"Internal error: no successful PyReadonlyArray2 extraction.",
))
};
match try_cast {
Ok(density_matrix) => Ok(Self {
internal: PragmaSetDensityMatrix::new(density_matrix),
}),
Err(_) => {
let density_matrix_casted: Vec<Vec<Complex64>> =
Vec::extract(density_matrix.as_borrowed())?;
let ncol = density_matrix_casted.first().map_or(0, |row| row.len());
let mut density_matrix_array2: Array2<Complex64> = Array2::zeros((0, ncol));
for subvec in density_matrix_casted {
let int_array1: Array1<Complex64> = Array1::from(subvec);
density_matrix_array2
.push_row((&int_array1).into())
.unwrap();
}
Ok(Self {
internal: PragmaSetDensityMatrix::new(density_matrix_array2),
})
}
}
}
fn density_matrix(&self) -> Py<PyArray2<Complex64>> {
Python::attach(|py| -> Py<PyArray2<Complex64>> {
self.internal.density_matrix().to_pyarray(py).unbind()
})
}
fn involved_qubits<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PySet>> {
PySet::new(py, ["All"])?
.into_pyobject(py)
.map_err(|_| PyRuntimeError::new_err("Unable to convert to Python object"))
}
fn tags(&self) -> Vec<String> {
self.internal.tags().iter().map(|s| s.to_string()).collect()
}
fn hqslang(&self) -> &'static str {
self.internal.hqslang()
}
fn is_parametrized(&self) -> bool {
self.internal.is_parametrized()
}
fn substitute_parameters(
&self,
substitution_parameters: std::collections::HashMap<String, f64>,
) -> PyResult<Self> {
let mut calculator = qoqo_calculator::Calculator::new();
for (key, val) in substitution_parameters.iter() {
calculator.set_variable(key, *val);
}
Ok(Self {
internal: self
.internal
.substitute_parameters(&calculator)
.map_err(|x| {
pyo3::exceptions::PyRuntimeError::new_err(format!(
"Parameter Substitution failed: {x:?}"
))
})?,
})
}
fn remap_qubits(&self, mapping: std::collections::HashMap<usize, usize>) -> PyResult<Self> {
let new_internal = self
.internal
.remap_qubits(&mapping)
.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Qubit remapping failed: "))?;
Ok(Self {
internal: new_internal,
})
}
fn __copy__(&self) -> PragmaSetDensityMatrixWrapper {
self.clone()
}
fn __deepcopy__(&self, _memodict: &Bound<PyAny>) -> PragmaSetDensityMatrixWrapper {
self.clone()
}
fn __format__(&self, _format_spec: &str) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __repr__(&self) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __richcmp__(
&self,
other: &Bound<PyAny>,
op: pyo3::class::basic::CompareOp,
) -> PyResult<bool> {
let other = crate::operations::convert_pyany_to_operation(other).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Right hand side cannot be converted to Operation",
)
})?;
match op {
pyo3::class::basic::CompareOp::Eq => {
Ok(Operation::from(self.internal.clone()) == other)
}
pyo3::class::basic::CompareOp::Ne => {
Ok(Operation::from(self.internal.clone()) != other)
}
_ => Err(pyo3::exceptions::PyNotImplementedError::new_err(
"Other comparison not implemented.",
)),
}
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn json_schema() -> String {
let schema = schemars::schema_for!(PragmaSetDensityMatrix);
serde_json::to_string_pretty(&schema).expect("Unexpected failure to serialize schema")
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn current_version() -> String {
ROQOQO_VERSION.to_string()
}
#[cfg(feature = "json_schema")]
pub fn min_supported_version(&self) -> String {
let min_version: (u32, u32, u32) =
PragmaSetDensityMatrix::minimum_supported_roqoqo_version(&self.internal);
format!("{}.{}.{}", min_version.0, min_version.1, min_version.2)
}
}
#[wrap(Operate, OperatePragma, JsonSchema)]
#[derive(Eq)]
struct PragmaRepeatGate {
repetition_coefficient: usize,
}
#[wrap(Operate, OperatePragma, OperateMultiQubit, JsonSchema)]
struct PragmaOverrotation {
gate_hqslang: String,
qubits: Vec<usize>,
amplitude: f64,
variance: f64,
}
#[wrap(Operate, OperatePragma, JsonSchema)]
struct PragmaBoostNoise {
noise_coefficient: CalculatorFloat,
}
#[wrap(Operate, OperateMultiQubit, OperatePragma, JsonSchema)]
struct PragmaStopParallelBlock {
qubits: Vec<usize>,
execution_time: CalculatorFloat,
}
#[wrap(Operate, JsonSchema)]
struct PragmaGlobalPhase {
phase: CalculatorFloat,
}
#[wrap(Operate, OperateMultiQubit, OperatePragma, JsonSchema)]
pub struct PragmaSleep {
qubits: Vec<usize>,
sleep_time: CalculatorFloat,
}
#[wrap(Operate, OperateSingleQubit, OperatePragma, JsonSchema)]
#[derive(Eq)]
pub struct PragmaActiveReset {
qubit: usize,
}
#[wrap(Operate, OperateMultiQubit, OperatePragma, JsonSchema)]
#[derive(Eq)]
pub struct PragmaStartDecompositionBlock {
qubits: Vec<usize>,
reordering_dictionary: HashMap<usize, usize>,
}
#[wrap(Operate, OperateMultiQubit, OperatePragma, JsonSchema)]
#[derive(Eq)]
pub struct PragmaStopDecompositionBlock {
qubits: Vec<usize>,
}
#[wrap(
Operate,
OperateSingleQubit,
OperatePragma,
OperatePragmaNoise,
OperatePragmaNoiseProba,
JsonSchema
)]
pub struct PragmaDamping {
qubit: usize,
gate_time: CalculatorFloat,
rate: CalculatorFloat,
}
#[wrap(
Operate,
OperateSingleQubit,
OperatePragma,
OperatePragmaNoise,
OperatePragmaNoiseProba,
JsonSchema
)]
pub struct PragmaDepolarising {
qubit: usize,
gate_time: CalculatorFloat,
rate: CalculatorFloat,
}
#[wrap(
Operate,
OperateSingleQubit,
OperatePragma,
OperatePragmaNoise,
OperatePragmaNoiseProba,
JsonSchema
)]
pub struct PragmaDephasing {
qubit: usize,
gate_time: CalculatorFloat,
rate: CalculatorFloat,
}
#[wrap(
Operate,
OperateSingleQubit,
OperatePragma,
OperatePragmaNoise,
OperatePragmaNoiseProba,
JsonSchema
)]
pub struct PragmaRandomNoise {
qubit: usize,
gate_time: CalculatorFloat,
depolarising_rate: CalculatorFloat,
dephasing_rate: CalculatorFloat,
}
#[pymodule]
fn pragma_general_noise(_py: Python, module: &Bound<PyModule>) -> PyResult<()> {
module.add_class::<PragmaGeneralNoiseWrapper>()?;
Ok(())
}
#[pyclass(
from_py_object,
name = "PragmaGeneralNoise",
module = "qoqo.operations"
)]
#[derive(Clone, Debug, PartialEq)]
pub struct PragmaGeneralNoiseWrapper {
pub internal: PragmaGeneralNoise,
}
insert_pyany_to_operation!(
"PragmaGeneralNoise" =>{
let qbt = op.call_method0("qubit")
.map_err(|_| QoqoError::ConversionError)?;
let qubit: usize = qbt.extract()
.map_err(|_| QoqoError::ConversionError)?;
let gatetm = &op.call_method0("gate_time")
.map_err(|_| QoqoError::ConversionError)?;
let gate_time: CalculatorFloat = convert_into_calculator_float(gatetm).map_err(|_| {
QoqoError::ConversionError
})?;
let array = op.call_method0("rates")
.map_err(|_| QoqoError::ConversionError)?;
let rates_array: PyReadonlyArray2<f64> = array.extract().unwrap();
let rates: Array2<f64> = rates_array.as_array().to_owned();
Ok(PragmaGeneralNoise::new(qubit, gate_time, rates).into())
}
);
insert_operation_to_pyobject!(
Operation::PragmaGeneralNoise(internal) => {
{
let pyref: Py<PragmaGeneralNoiseWrapper> =
Py::new(py, PragmaGeneralNoiseWrapper { internal }).unwrap();
pyref.into_pyobject(py).map(|bound| bound.as_any().to_owned()).map_err(|_| PyValueError::new_err("Unable to convert to Python object"))
}
}
);
#[pymethods]
impl PragmaGeneralNoiseWrapper {
#[new]
fn new(qubit: usize, gate_time: &Bound<PyAny>, rates: &Bound<PyAny>) -> PyResult<Self> {
let rates_array: Array2<f64> =
if let Ok(rates_pyarray) = rates.extract::<PyReadonlyArray2<f64>>() {
rates_pyarray.as_array().to_owned()
} else {
let rates_casted: Vec<Vec<f64>> = Vec::extract(rates.as_borrowed())?;
let ncol = rates_casted.first().map_or(0, |row| row.len());
let mut rates_array2: Array2<f64> = Array2::zeros((0, ncol));
for subvec in rates_casted {
let int_array1: Array1<f64> = Array1::from(subvec);
rates_array2.push_row((&int_array1).into()).unwrap();
}
rates_array2
};
let gate_time_cf = convert_into_calculator_float(gate_time).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Argument gate time cannot be converted to CalculatorFloat",
)
})?;
Ok(Self {
internal: PragmaGeneralNoise::new(qubit, gate_time_cf, rates_array),
})
}
fn qubit(&self) -> usize {
*self.internal.qubit()
}
fn gate_time(&self) -> CalculatorFloatWrapper {
CalculatorFloatWrapper {
internal: self.internal.gate_time().clone(),
}
}
fn rates(&self) -> Py<PyArray2<f64>> {
Python::attach(|py| -> Py<PyArray2<f64>> { self.internal.rates().to_pyarray(py).unbind() })
}
fn superoperator(&self) -> PyResult<Py<PyArray2<f64>>> {
Python::attach(|py| -> PyResult<Py<PyArray2<f64>>> {
match self.internal.superoperator() {
Ok(x) => Ok(x.to_pyarray(py).unbind()),
Err(err) => Err(PyRuntimeError::new_err(format!("{err:?}"))),
}
})
}
fn involved_qubits<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PySet>> {
PySet::new(py, [*self.internal.qubit()])?
.into_pyobject(py)
.map_err(|_| PyRuntimeError::new_err("Unable to convert to Python object"))
}
fn tags(&self) -> Vec<String> {
self.internal.tags().iter().map(|s| s.to_string()).collect()
}
fn hqslang(&self) -> &'static str {
self.internal.hqslang()
}
fn is_parametrized(&self) -> bool {
self.internal.is_parametrized()
}
fn substitute_parameters(
&self,
substitution_parameters: std::collections::HashMap<String, f64>,
) -> PyResult<Self> {
let mut calculator = qoqo_calculator::Calculator::new();
for (key, val) in substitution_parameters.iter() {
calculator.set_variable(key, *val);
}
Ok(Self {
internal: self
.internal
.substitute_parameters(&calculator)
.map_err(|x| {
pyo3::exceptions::PyRuntimeError::new_err(format!(
"Parameter Substitution failed: {x:?}"
))
})?,
})
}
fn remap_qubits(&self, mapping: std::collections::HashMap<usize, usize>) -> PyResult<Self> {
let new_internal = self
.internal
.remap_qubits(&mapping)
.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Qubit remapping failed: "))?;
Ok(Self {
internal: new_internal,
})
}
fn __copy__(&self) -> PragmaGeneralNoiseWrapper {
self.clone()
}
fn __deepcopy__(&self, _memodict: &Bound<PyAny>) -> PragmaGeneralNoiseWrapper {
self.clone()
}
fn __format__(&self, _format_spec: &str) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __repr__(&self) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __richcmp__(
&self,
other: &Bound<PyAny>,
op: pyo3::class::basic::CompareOp,
) -> PyResult<bool> {
let other = crate::operations::convert_pyany_to_operation(other).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Right hand side cannot be converted to Operation",
)
})?;
match op {
pyo3::class::basic::CompareOp::Eq => {
Ok(Operation::from(self.internal.clone()) == other)
}
pyo3::class::basic::CompareOp::Ne => {
Ok(Operation::from(self.internal.clone()) != other)
}
_ => Err(pyo3::exceptions::PyNotImplementedError::new_err(
"Other comparison not implemented.",
)),
}
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn json_schema() -> String {
let schema = schemars::schema_for!(PragmaGeneralNoise);
serde_json::to_string_pretty(&schema).expect("Unexpected failure to serialize schema")
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn current_version() -> String {
ROQOQO_VERSION.to_string()
}
#[cfg(feature = "json_schema")]
pub fn min_supported_version(&self) -> String {
let min_version: (u32, u32, u32) =
PragmaGeneralNoise::minimum_supported_roqoqo_version(&self.internal);
format!("{}.{}.{}", min_version.0, min_version.1, min_version.2)
}
}
#[wrap(Operate, OperatePragma, JsonSchema)]
pub struct PragmaConditional {
condition_register: String,
condition_index: usize,
circuit: Circuit,
}
#[wrap(Operate, OperatePragma, JsonSchema)]
pub struct PragmaControlledCircuit {
controlling_qubit: usize,
circuit: Circuit,
}
#[pyclass(
from_py_object,
name = "PragmaChangeDevice",
module = "qoqo.operations"
)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PragmaChangeDeviceWrapper {
pub internal: PragmaChangeDevice,
}
insert_pyany_to_operation!(
"PragmaChangeDevice" =>{
let wt = op.call_method0( "wrapped_tags").map_err(|_|QoqoError::ConversionError)?;
let wrapped_tags: Vec<String> = wt.extract()
.map_err(|_| QoqoError::ConversionError)?;
let wh = op.call_method0( "wrapped_hqslang").map_err(|_|QoqoError::ConversionError)?;
let wrapped_hqslang: String = wh.extract()
.map_err(|_|QoqoError::ConversionError)?;
let wo = op.call_method0( "wrapped_operation").map_err(|_|QoqoError::ConversionError)?;
let wrapped_operation: Vec<u8> = wo.extract()
.map_err(|_|QoqoError::ConversionError)?;
Ok( PragmaChangeDevice{wrapped_tags, wrapped_hqslang, wrapped_operation}.into())
}
);
insert_operation_to_pyobject!(
Operation::PragmaChangeDevice(internal) => {
{
let pyref: Py<PragmaChangeDeviceWrapper> =
Py::new(py, PragmaChangeDeviceWrapper { internal }).unwrap();
pyref.into_pyobject(py).map(|bound| bound.as_any().to_owned()).map_err(|_| PyValueError::new_err("Unable to convert to Python object"))
}
}
);
#[pymethods]
impl PragmaChangeDeviceWrapper {
#[new]
fn new() -> PyResult<Self> {
Err(PyTypeError::new_err("A PragmaChangeDevice wrapper Pragma cannot be created directly, use a .to_pragma_change_device() from the wrapped PRAGMA instead"))
}
fn wrapped_tags(&self) -> Vec<String> {
self.internal
.wrapped_tags
.iter()
.map(|s| s.to_string())
.collect()
}
fn wrapped_hqslang(&self) -> String {
self.internal.wrapped_hqslang.to_string()
}
fn wrapped_operation(&self) -> PyResult<Py<PyByteArray>> {
let serialized: Vec<u8> = self.internal.wrapped_operation.clone();
let b: Py<PyByteArray> = Python::attach(|py| -> Py<PyByteArray> {
PyByteArray::new(py, &serialized[..]).into()
});
Ok(b)
}
fn involved_qubits<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PySet>> {
PySet::new(py, ["All"])?
.into_pyobject(py)
.map_err(|_| PyRuntimeError::new_err("Unable to convert to Python object"))
}
fn tags(&self) -> Vec<String> {
self.internal.tags().iter().map(|s| s.to_string()).collect()
}
fn hqslang(&self) -> &'static str {
self.internal.hqslang()
}
fn is_parametrized(&self) -> bool {
self.internal.is_parametrized()
}
fn substitute_parameters(
&self,
substitution_parameters: std::collections::HashMap<String, f64>,
) -> PyResult<Self> {
let mut calculator = qoqo_calculator::Calculator::new();
for (key, val) in substitution_parameters.iter() {
calculator.set_variable(key, *val);
}
Ok(Self {
internal: self
.internal
.substitute_parameters(&calculator)
.map_err(|x| {
pyo3::exceptions::PyRuntimeError::new_err(format!(
"Parameter Substitution failed: {x:?}"
))
})?,
})
}
fn remap_qubits(&self, mapping: std::collections::HashMap<usize, usize>) -> PyResult<Self> {
let new_internal = self
.internal
.remap_qubits(&mapping)
.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Qubit remapping failed: "))?;
Ok(Self {
internal: new_internal,
})
}
fn __copy__(&self) -> PragmaChangeDeviceWrapper {
self.clone()
}
fn __deepcopy__(&self, _memodict: &Bound<PyAny>) -> PragmaChangeDeviceWrapper {
self.clone()
}
fn __format__(&self, _format_spec: &str) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __repr__(&self) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __richcmp__(
&self,
other: &Bound<PyAny>,
op: pyo3::class::basic::CompareOp,
) -> PyResult<bool> {
let other = crate::operations::convert_pyany_to_operation(other).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Right hand side cannot be converted to Operation",
)
})?;
match op {
pyo3::class::basic::CompareOp::Eq => {
Ok(Operation::from(self.internal.clone()) == other)
}
pyo3::class::basic::CompareOp::Ne => {
Ok(Operation::from(self.internal.clone()) != other)
}
_ => Err(pyo3::exceptions::PyNotImplementedError::new_err(
"Other comparison not implemented.",
)),
}
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn json_schema() -> String {
let schema = schemars::schema_for!(PragmaChangeDevice);
serde_json::to_string_pretty(&schema).expect("Unexpected failure to serialize schema")
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn current_version() -> String {
ROQOQO_VERSION.to_string()
}
#[cfg(feature = "json_schema")]
pub fn min_supported_version(&self) -> String {
let min_version: (u32, u32, u32) =
PragmaChangeDevice::minimum_supported_roqoqo_version(&self.internal);
format!("{}.{}.{}", min_version.0, min_version.1, min_version.2)
}
}
#[pymodule]
fn pragma_annotated_op(_py: Python, module: &Bound<PyModule>) -> PyResult<()> {
module.add_class::<PragmaAnnotatedOpWrapper>()?;
Ok(())
}
#[pyclass(from_py_object, name = "PragmaAnnotatedOp", module = "qoqo.operations")]
#[derive(Clone, Debug, PartialEq)]
pub struct PragmaAnnotatedOpWrapper {
pub internal: PragmaAnnotatedOp,
}
insert_pyany_to_operation!(
"PragmaAnnotatedOp" =>{
let annot_op = &op.call_method0( "operation").map_err(|_|QoqoError::ConversionError)?;
let operation: Operation = convert_pyany_to_operation(annot_op)
.map_err(|_| QoqoError::ConversionError)?;
let annot = op.call_method0( "annotation").map_err(|_|QoqoError::ConversionError)?;
let annotation: String = annot.extract()
.map_err(|_|QoqoError::ConversionError)?;
Ok( PragmaAnnotatedOp{ operation: Box::new(operation), annotation }.into())
}
);
insert_operation_to_pyobject!(
Operation::PragmaAnnotatedOp(internal) => {
{
let pyref: Py<PragmaAnnotatedOpWrapper> =
Py::new(py, PragmaAnnotatedOpWrapper { internal }).unwrap();
pyref.into_pyobject(py).map(|bound| bound.as_any().to_owned()).map_err(|_| PyValueError::new_err("Unable to convert to Python object"))
}
}
);
#[pymethods]
impl PragmaAnnotatedOpWrapper {
#[new]
fn new(operation: &Bound<PyAny>, annotation: String) -> PyResult<Self> {
let op = crate::operations::convert_pyany_to_operation(operation).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Input operation cannot be converted to Operation",
)
})?;
Ok(Self {
internal: PragmaAnnotatedOp::new(op, annotation),
})
}
fn operation<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let op = self.internal.operation.clone();
convert_operation_to_pyobject(*op, py)
}
fn annotation(&self) -> String {
self.internal.annotation.clone()
}
fn involved_qubits<'py>(&'py self, py: Python<'py>) -> Bound<'py, PySet> {
let involved = self.internal.involved_qubits();
match involved {
InvolvedQubits::All => PySet::new(py, ["All"]).expect("Couldn not create PySet"),
InvolvedQubits::None => PySet::empty(py).expect("Couldn not create PySet"),
InvolvedQubits::Set(x) => {
let mut vector: Vec<usize> = Vec::new();
for qubit in x {
vector.push(qubit)
}
PySet::new(py, &vector[..]).expect("Couldn not create PySet")
}
}
}
fn tags(&self) -> Vec<String> {
self.internal.tags().iter().map(|s| s.to_string()).collect()
}
fn hqslang(&self) -> &'static str {
self.internal.hqslang()
}
fn is_parametrized(&self) -> bool {
self.internal.is_parametrized()
}
fn substitute_parameters(
&self,
substitution_parameters: std::collections::HashMap<String, f64>,
) -> PyResult<Self> {
let mut calculator = qoqo_calculator::Calculator::new();
for (key, val) in substitution_parameters.iter() {
calculator.set_variable(key, *val);
}
Ok(Self {
internal: self
.internal
.substitute_parameters(&calculator)
.map_err(|x| {
pyo3::exceptions::PyRuntimeError::new_err(format!(
"Parameter Substitution failed: {x:?}"
))
})?,
})
}
fn remap_qubits(&self, mapping: std::collections::HashMap<usize, usize>) -> PyResult<Self> {
let new_internal = self
.internal
.remap_qubits(&mapping)
.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Qubit remapping failed: "))?;
Ok(Self {
internal: new_internal,
})
}
fn __copy__(&self) -> PragmaAnnotatedOpWrapper {
self.clone()
}
fn __deepcopy__(&self, _memodict: &Bound<PyAny>) -> PragmaAnnotatedOpWrapper {
self.clone()
}
fn __format__(&self, _format_spec: &str) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __repr__(&self) -> PyResult<String> {
Ok(format!("{:?}", self.internal))
}
fn __richcmp__(
&self,
other: &Bound<PyAny>,
op: pyo3::class::basic::CompareOp,
) -> PyResult<bool> {
let other = crate::operations::convert_pyany_to_operation(other).map_err(|_| {
pyo3::exceptions::PyTypeError::new_err(
"Right hand side cannot be converted to Operation",
)
})?;
match op {
pyo3::class::basic::CompareOp::Eq => {
Ok(Operation::from(self.internal.clone()) == other)
}
pyo3::class::basic::CompareOp::Ne => {
Ok(Operation::from(self.internal.clone()) != other)
}
_ => Err(pyo3::exceptions::PyNotImplementedError::new_err(
"Other comparison not implemented.",
)),
}
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn json_schema() -> String {
let schema = schemars::schema_for!(PragmaAnnotatedOp);
serde_json::to_string_pretty(&schema).expect("Unexpected failure to serialize schema")
}
#[cfg(feature = "json_schema")]
#[staticmethod]
pub fn current_version() -> String {
ROQOQO_VERSION.to_string()
}
#[cfg(feature = "json_schema")]
pub fn min_supported_version(&self) -> String {
let min_version: (u32, u32, u32) =
PragmaAnnotatedOp::minimum_supported_roqoqo_version(&self.internal);
format!("{}.{}.{}", min_version.0, min_version.1, min_version.2)
}
}
#[cfg(test)]
mod tests {
use crate::operations::*;
use roqoqo::operations::*;
use std::collections::HashSet;
#[test]
fn test_pyo3_involved_qubits_all_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_definition: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_definition, py).unwrap();
let to_involved = operation.call_method0("involved_qubits").unwrap();
let involved_op: HashSet<String> = HashSet::extract(to_involved.as_borrowed()).unwrap();
let mut involved_param: HashSet<String> = HashSet::new();
involved_param.insert("All".to_owned());
assert_eq!(involved_op, involved_param);
assert!(PragmaChangeDeviceWrapper::new().is_err());
})
}
#[test]
fn test_pyo3_format_repr_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_measurement: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
let format_repr = format!("PragmaChangeDevice {{ wrapped_tags: {:?}, wrapped_hqslang: {:?}, wrapped_operation: {:?} }}", wrapped.tags(), wrapped.hqslang(), bincode::serde::encode_to_vec(&wrapped, bincode::config::legacy()).unwrap());
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_measurement, py).unwrap();
let to_format = operation.call_method1("__format__", ("",)).unwrap();
let format_op: String = String::extract(to_format.as_borrowed()).unwrap();
let to_repr = operation.call_method0("__repr__").unwrap();
let repr_op: String = String::extract(to_repr.as_borrowed()).unwrap();
assert_eq!(format_op, format_repr);
assert_eq!(repr_op, format_repr);
})
}
#[test]
fn test_pyo3_copy_deepcopy_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_measurement: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_measurement, py).unwrap();
let copy_op = operation.call_method0("__copy__").unwrap();
let deepcopy_op = operation.call_method1("__deepcopy__", ("",)).unwrap();
let copy_deepcopy_param = operation;
let comparison_copy = bool::extract(
copy_op
.call_method1("__eq__", (copy_deepcopy_param.clone(),))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(comparison_copy);
let comparison_deepcopy = bool::extract(
deepcopy_op
.call_method1("__eq__", (copy_deepcopy_param,))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(comparison_deepcopy);
})
}
#[test]
fn test_pyo3_tags_simple_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_measurement: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_measurement, py).unwrap();
let to_tag = operation.call_method0("tags").unwrap();
let tags_op: &Vec<String> = &Vec::extract(to_tag.as_borrowed()).unwrap();
let tags_param: &[&str] = &["Operation", "PragmaOperation", "PragmaChangeDevice"];
assert_eq!(tags_op, tags_param);
})
}
#[test]
fn test_pyo3_hqslang_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_measurement: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_measurement, py).unwrap();
let hqslang_op: String =
String::extract(operation.call_method0("hqslang").unwrap().as_borrowed()).unwrap();
assert_eq!(hqslang_op, "PragmaChangeDevice".to_string());
})
}
#[test]
fn test_pyo3_is_parametrized_change_device() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let input_measurement: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(input_measurement, py).unwrap();
assert!(!bool::extract(
operation
.call_method0("is_parametrized")
.unwrap()
.as_borrowed()
)
.unwrap());
})
}
#[test]
fn test_pyo3_substitute_parameters() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let first_op: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
let second_op: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(first_op, py).unwrap();
let mut substitution_dict: HashMap<String, f64> = HashMap::new();
substitution_dict.insert("test".to_owned(), 1.0);
let substitute_op = operation
.call_method1("substitute_parameters", (substitution_dict,))
.unwrap();
let substitute_param = convert_operation_to_pyobject(second_op, py).unwrap();
let comparison = bool::extract(
substitute_op
.call_method1("__eq__", (substitute_param,))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(comparison);
})
}
#[test]
fn test_pyo3_remap_qubits() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let first_op: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
let second_op: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(first_op, py).unwrap();
let mut qubit_mapping: HashMap<usize, usize> = HashMap::new();
qubit_mapping.insert(0, 0);
let remapped_op = operation
.call_method1("remap_qubits", (qubit_mapping,))
.unwrap();
let comparison_op = convert_operation_to_pyobject(second_op, py).unwrap();
let comparison = bool::extract(
remapped_op
.call_method1("__eq__", (comparison_op,))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(comparison);
})
}
#[test]
fn test_pyo3_remap_qubits_error() {
let wrapped: Operation = PragmaActiveReset::new(0).into();
let first_op: Operation = PragmaChangeDevice::new(&wrapped).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation = convert_operation_to_pyobject(first_op, py).unwrap();
let mut qubit_mapping: HashMap<usize, usize> = HashMap::new();
qubit_mapping.insert(0, 2);
let remapped_op = operation.call_method1("remap_qubits", (qubit_mapping,));
assert!(remapped_op.is_err());
})
}
#[test]
fn test_pyo3_richcmp_change_device() {
let wrapped_1: Operation = PragmaActiveReset::new(0).into();
let definition_1: Operation = PragmaChangeDevice::new(&wrapped_1).unwrap().into();
let wrapped_2: Operation = PragmaActiveReset::new(1).into();
let definition_2: Operation = PragmaChangeDevice::new(&wrapped_2).unwrap().into();
Python::initialize();
Python::attach(|py| {
let operation_one = convert_operation_to_pyobject(definition_1, py).unwrap();
let operation_two = convert_operation_to_pyobject(definition_2, py).unwrap();
let comparison = bool::extract(
operation_one
.call_method1("__eq__", (operation_two.clone(),))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(!comparison);
let comparison = bool::extract(
operation_one
.call_method1("__ne__", (operation_two.clone(),))
.unwrap()
.as_borrowed(),
)
.unwrap();
assert!(comparison);
let comparison = operation_one.call_method1("__eq__", (vec!["fails"],));
assert!(comparison.is_err());
let comparison = operation_one.call_method1("__ge__", (operation_two,));
assert!(comparison.is_err());
})
}
}