1use opentelemetry::trace::TraceContextExt;
2use qcs::compiler::quilc::{
3 CompilerOpts, ConjugateByCliffordRequest, ConjugatePauliByCliffordResponse,
4 GenerateRandomizedBenchmarkingSequenceResponse, NativeQuilMetadata, PauliTerm,
5 RandomizedBenchmarkingRequest, TargetDevice, DEFAULT_COMPILER_TIMEOUT,
6};
7use qcs_api_client_openapi::models::InstructionSetArchitecture;
8use quil_rs::quil::Quil;
9use rigetti_pyo3::{
10 create_init_submodule, impl_repr, py_wrap_data_struct, py_wrap_error, py_wrap_struct,
11 py_wrap_type,
12 pyo3::{
13 exceptions::{PyRuntimeError, PyValueError},
14 pyclass, pyfunction, pymethods,
15 types::{PyBytes, PyFloat, PyInt, PyString},
16 Py, PyResult, Python,
17 },
18 wrap_error, PyWrapper, ToPythonError,
19};
20
21use crate::qpu::isa::PyInstructionSetArchitecture;
22
23create_init_submodule! {
24 classes: [
25 PyCompilerOpts,
26 PyCompilationResult,
27 PyNativeQuilMetadata,
28 PyTargetDevice,
29 PyPauliTerm,
30 PyConjugateByCliffordRequest,
31 PyConjugatePauliByCliffordResponse,
32 PyRandomizedBenchmarkingRequest,
33 PyGenerateRandomizedBenchmarkingSequenceResponse,
34 PyQuilcClient
35 ],
36 consts: [DEFAULT_COMPILER_TIMEOUT],
37 errors: [QuilcError],
38 funcs: [
39 py_compile_program,
40 py_compile_program_async,
41 py_get_version_info,
42 py_get_version_info_async,
43 py_conjugate_pauli_by_clifford,
44 py_conjugate_pauli_by_clifford_async,
45 py_generate_randomized_benchmarking_sequence,
46 py_generate_randomized_benchmarking_sequence_async
47 ],
48}
49
50py_wrap_type! {
51 #[derive(Default)]
52 PyCompilerOpts(CompilerOpts) as "CompilerOpts";
53}
54
55#[pymethods]
56impl PyCompilerOpts {
57 #[new]
58 #[pyo3(signature = (/, timeout = DEFAULT_COMPILER_TIMEOUT, protoquil = None))]
59 pub fn new(timeout: Option<f64>, protoquil: Option<bool>) -> Self {
60 let opts = CompilerOpts::new()
61 .with_timeout(timeout)
62 .with_protoquil(protoquil);
63 Self(opts)
64 }
65
66 #[staticmethod]
67 #[allow(clippy::should_implement_trait)]
68 pub fn default() -> Self {
69 <Self as Default>::default()
70 }
71}
72
73wrap_error!(RustQuilcError(qcs::compiler::quilc::Error));
74py_wrap_error!(quilc, RustQuilcError, QuilcError, PyRuntimeError);
75
76py_wrap_struct! {
77 PyTargetDevice(TargetDevice) as "TargetDevice" {}
78}
79
80#[pymethods]
81impl PyTargetDevice {
82 #[staticmethod]
83 pub fn from_isa(isa: PyInstructionSetArchitecture) -> PyResult<Self> {
84 let isa: InstructionSetArchitecture = isa.into();
85 let target: TargetDevice = isa
86 .try_into()
87 .map_err(RustQuilcError::from)
88 .map_err(RustQuilcError::to_py_err)?;
89
90 Ok(Self(target))
91 }
92
93 #[staticmethod]
94 pub fn from_json(value: String) -> PyResult<Self> {
95 let target: TargetDevice = serde_json::from_str(&value)
96 .map_err(|err| err.to_string())
97 .map_err(PyValueError::new_err)?;
98
99 Ok(Self(target))
100 }
101}
102
103#[derive(Clone, Debug)]
104pub enum QuilcClient {
105 Rpcq(qcs::compiler::rpcq::Client),
106 #[cfg(feature = "libquil")]
107 LibquilSys(qcs::compiler::libquil::Client),
108}
109
110impl QuilcClient {
111 pub(crate) fn as_client(&self) -> &dyn qcs::compiler::quilc::Client {
112 match self {
113 QuilcClient::Rpcq(client) => client,
114 #[cfg(feature = "libquil")]
115 QuilcClient::LibquilSys(client) => client,
116 }
117 }
118}
119
120impl qcs::compiler::quilc::Client for QuilcClient {
121 fn compile_program(
122 &self,
123 quil: &str,
124 isa: TargetDevice,
125 options: CompilerOpts,
126 ) -> Result<qcs::compiler::quilc::CompilationResult, qcs::compiler::quilc::Error> {
127 self.as_client().compile_program(quil, isa, options)
128 }
129
130 fn get_version_info(&self) -> Result<String, qcs::compiler::quilc::Error> {
131 self.as_client().get_version_info()
132 }
133
134 fn conjugate_pauli_by_clifford(
135 &self,
136 request: ConjugateByCliffordRequest,
137 ) -> Result<ConjugatePauliByCliffordResponse, qcs::compiler::quilc::Error> {
138 self.as_client().conjugate_pauli_by_clifford(request)
139 }
140
141 fn generate_randomized_benchmarking_sequence(
142 &self,
143 request: RandomizedBenchmarkingRequest,
144 ) -> Result<GenerateRandomizedBenchmarkingSequenceResponse, qcs::compiler::quilc::Error> {
145 self.as_client()
146 .generate_randomized_benchmarking_sequence(request)
147 }
148}
149
150#[derive(Debug, thiserror::Error)]
151pub enum Error {
152 #[error("error when compiling with RPCQ client: {0}")]
153 Rpcq(#[from] qcs::compiler::quilc::Error),
154 #[cfg(feature = "libquil")]
155 #[error("error when compiling with libquil client: {0}")]
156 Libquil(#[from] qcs::compiler::libquil::Error),
157}
158
159#[pyclass(name = "QuilcClient")]
160#[derive(Clone)]
161pub struct PyQuilcClient {
162 pub inner: QuilcClient,
163}
164
165#[pymethods]
166impl PyQuilcClient {
167 #[new]
168 fn new() -> PyResult<Self> {
169 Err(PyRuntimeError::new_err(
170 "QuilcClient cannot not be instantiated directly. See the static methods: QuilcClient.new_rpcq().",
171 ))
172 }
173
174 #[staticmethod]
175 fn new_rpcq(endpoint: &str) -> PyResult<Self> {
176 let rpcq_client = qcs::compiler::rpcq::Client::new(endpoint)
177 .map_err(|e| qcs::compiler::quilc::Error::QuilcConnection(endpoint.into(), e))
178 .map_err(RustQuilcError::from)
179 .map_err(RustQuilcError::to_py_err)?;
180 Ok(Self {
181 inner: QuilcClient::Rpcq(rpcq_client),
182 })
183 }
184
185 #[cfg(feature = "libquil")]
186 #[staticmethod]
187 fn new_libquil() -> PyResult<Self> {
188 let libquil_client = qcs::compiler::libquil::Client {};
189 Ok(Self {
190 inner: QuilcClient::LibquilSys(libquil_client),
191 })
192 }
193}
194
195crate::py_sync::py_function_sync_async! {
196 #[pyfunction]
197 #[pyo3(signature = (quil, target, client, options = None))]
198 #[tracing::instrument(skip_all)]
199 #[pyo3_opentelemetry::pypropagate(on_context_extraction_failure="ignore")]
200 async fn compile_program(
201 quil: String,
202 target: PyTargetDevice,
203 client: PyQuilcClient,
204 options: Option<PyCompilerOpts>,
205 ) -> PyResult<PyCompilationResult> {
206 let client = client.inner.as_client();
207 let options = options.unwrap_or_default();
208 client.compile_program(&quil, target.into(), options.into())
209 .map_err(RustQuilcError::from)
210 .map_err(RustQuilcError::to_py_err)
211 .map(|result| PyCompilationResult {
212 program: result.program.to_quil().expect("successfully compiled program should convert to valid quil"),
213 native_quil_metadata: result.native_quil_metadata.map(PyNativeQuilMetadata)
214 })
215
216 }
217}
218
219py_wrap_data_struct! {
220 #[derive(Debug, PartialEq, PartialOrd)]
221 PyNativeQuilMetadata(NativeQuilMetadata) as "NativeQuilMetadata" {
222 final_rewiring: Vec<u64> => Vec<Py<PyInt>>,
223 gate_depth: Option<u64> => Option<Py<PyInt>>,
224 gate_volume: Option<u64> => Option<Py<PyInt>>,
225 multiqubit_gate_depth: Option<u64> => Option<Py<PyInt>>,
226 program_duration: Option<f64> => Option<Py<PyFloat>>,
227 program_fidelity: Option<f64> => Option<Py<PyFloat>>,
228 topological_swaps: Option<u64> => Option<Py<PyInt>>,
229 qpu_runtime_estimation: Option<f64> => Option<Py<PyFloat>>
230
231 }
232}
233impl_repr!(PyNativeQuilMetadata);
234
235#[pymethods]
236impl PyNativeQuilMetadata {
237 #[new]
238 #[allow(clippy::too_many_arguments)]
239 pub fn new(
240 final_rewiring: Option<Vec<u64>>,
241 gate_depth: Option<u64>,
242 gate_volume: Option<u64>,
243 multiqubit_gate_depth: Option<u64>,
244 program_duration: Option<f64>,
245 program_fidelity: Option<f64>,
246 topological_swaps: Option<u64>,
247 qpu_runtime_estimation: Option<f64>,
248 ) -> Self {
249 Self(NativeQuilMetadata {
250 final_rewiring: final_rewiring.unwrap_or_default(),
251 gate_depth,
252 gate_volume,
253 multiqubit_gate_depth,
254 program_duration,
255 program_fidelity,
256 topological_swaps,
257 qpu_runtime_estimation,
258 })
259 }
260
261 pub fn __getstate__<'a>(&self, py: Python<'a>) -> PyResult<&'a PyBytes> {
262 Ok(PyBytes::new(
263 py,
264 &serde_json::to_vec(self.as_inner())
265 .map_err(|e| PyRuntimeError::new_err(format!("failed to serialize: {e}")))?,
266 ))
267 }
268
269 pub fn __setstate__(&mut self, state: &PyBytes) -> PyResult<()> {
270 let metadata: NativeQuilMetadata = serde_json::from_slice(state.as_bytes())
271 .map_err(|e| PyRuntimeError::new_err(format!("failed to deserialize: {e}")))?;
272 *self = Self(metadata);
273 Ok(())
274 }
275}
276
277#[pyclass(name = "CompilationResult")]
278pub struct PyCompilationResult {
279 #[pyo3(get)]
280 program: String,
281 #[pyo3(get)]
282 native_quil_metadata: Option<PyNativeQuilMetadata>,
283}
284
285crate::py_sync::py_function_sync_async! {
286 #[pyfunction]
287 async fn get_version_info(
288 client: PyQuilcClient,
289 ) -> PyResult<String> {
290 let client = client.inner.as_client();
291 client.get_version_info()
292 .map_err(RustQuilcError::from)
293 .map_err(RustQuilcError::to_py_err)
294 }
295}
296
297py_wrap_data_struct! {
298 PyPauliTerm(PauliTerm) as "PauliTerm" {
299 indices: Vec<u64> => Vec<Py<PyInt>>,
300 symbols: Vec<String> => Vec<Py<PyString>>
301 }
302}
303
304#[pymethods]
305impl PyPauliTerm {
306 #[new]
307 fn __new__(indices: Vec<u64>, symbols: Vec<String>) -> PyResult<Self> {
308 Ok(Self(PauliTerm { indices, symbols }))
309 }
310}
311
312py_wrap_data_struct! {
313 PyConjugateByCliffordRequest(ConjugateByCliffordRequest) as "ConjugateByCliffordRequest" {
314 pauli: PauliTerm => PyPauliTerm,
315 clifford: String => Py<PyString>
316 }
317}
318
319#[pymethods]
320impl PyConjugateByCliffordRequest {
321 #[new]
322 fn __new__(pauli: PyPauliTerm, clifford: String) -> PyResult<Self> {
323 Ok(Self(ConjugateByCliffordRequest {
324 pauli: pauli.into(),
325 clifford,
326 }))
327 }
328}
329
330py_wrap_data_struct! {
331 PyConjugatePauliByCliffordResponse(ConjugatePauliByCliffordResponse) as "ConjugatePauliByCliffordResponse" {
332 phase: i64 => Py<PyInt>,
333 pauli: String => Py<PyString>
334 }
335}
336
337crate::py_sync::py_function_sync_async! {
338 #[pyfunction]
339 async fn conjugate_pauli_by_clifford(
340 request: PyConjugateByCliffordRequest,
341 client: PyQuilcClient,
342 ) -> PyResult<PyConjugatePauliByCliffordResponse> {
343 let client = client.inner.as_client();
344 client.conjugate_pauli_by_clifford(request.into())
345 .map(PyConjugatePauliByCliffordResponse::from)
346 .map_err(RustQuilcError::from)
347 .map_err(RustQuilcError::to_py_err)
348 }
349}
350
351py_wrap_data_struct! {
352 PyRandomizedBenchmarkingRequest(RandomizedBenchmarkingRequest) as "RandomizedBenchmarkingRequest" {
353 depth: u64 => Py<PyInt>,
354 qubits: u64 => Py<PyInt>,
355 gateset: Vec<String> => Vec<Py<PyString>>,
356 seed: Option<u64> => Option<Py<PyInt>>,
357 interleaver: Option<String> => Option<Py<PyString>>
358 }
359}
360
361#[pymethods]
362impl PyRandomizedBenchmarkingRequest {
363 #[new]
364 fn __new__(
365 depth: u64,
366 qubits: u64,
367 gateset: Vec<String>,
368 seed: Option<u64>,
369 interleaver: Option<String>,
370 ) -> PyResult<Self> {
371 Ok(Self(RandomizedBenchmarkingRequest {
372 depth,
373 qubits,
374 gateset,
375 seed,
376 interleaver,
377 }))
378 }
379}
380
381py_wrap_data_struct! {
382 PyGenerateRandomizedBenchmarkingSequenceResponse(GenerateRandomizedBenchmarkingSequenceResponse) as "GenerateRandomizedBenchmarkingSequenceResponse" {
383 sequence: Vec<Vec<i64>> => Vec<Vec<Py<PyInt>>>
384 }
385}
386
387crate::py_sync::py_function_sync_async! {
388 #[pyfunction]
389 async fn generate_randomized_benchmarking_sequence(
390 request: PyRandomizedBenchmarkingRequest,
391 client: PyQuilcClient,
392 ) -> PyResult<PyGenerateRandomizedBenchmarkingSequenceResponse> {
393 let client = client.inner.as_client();
394 client.generate_randomized_benchmarking_sequence(request.into())
395 .map(PyGenerateRandomizedBenchmarkingSequenceResponse::from)
396 .map_err(RustQuilcError::from)
397 .map_err(RustQuilcError::to_py_err)
398 }
399}