qcs_sdk/qpu/
translation.rs

1//! Translating programs.
2use std::{collections::HashMap, time::Duration};
3
4use futures_util::TryFutureExt;
5use opentelemetry::trace::FutureExt;
6use prost::Message;
7use pyo3::types::PyBytes;
8use pyo3::Python;
9use pyo3::{exceptions::PyRuntimeError, pyclass, pyfunction, pymethods, PyResult};
10use qcs::qpu::translation::TranslationOptions;
11use qcs_api_client_grpc::services::translation::translation_options::{self, Riverlane};
12use qcs_api_client_grpc::services::translation::{
13    translation_options::TranslationBackend as ApiTranslationBackend,
14    TranslationOptions as ApiTranslationOptions,
15};
16use rigetti_pyo3::{create_init_submodule, py_wrap_error, py_wrap_simple_enum, ToPythonError};
17
18use crate::client::PyQcsClient;
19
20create_init_submodule! {
21    classes: [
22        PyTranslationOptions,
23        PyTranslationResult,
24        PyTranslationBackend,
25        PyQCtrl,
26        PyRiverlane
27    ],
28    errors: [
29        TranslationError
30    ],
31    funcs: [
32        py_get_quilt_calibrations,
33        py_get_quilt_calibrations_async,
34        py_translate,
35        py_translate_async
36    ],
37}
38
39crate::py_sync::py_function_sync_async! {
40    /// Query the QCS API for Quil-T calibrations.
41    /// If `None`, the default `timeout` used is 10 seconds.
42    #[pyfunction]
43    #[pyo3(signature = (quantum_processor_id, client = None, timeout = None))]
44    #[pyo3_opentelemetry::pypropagate(on_context_extraction_failure="ignore")]
45    async fn get_quilt_calibrations(
46        quantum_processor_id: String,
47        client: Option<PyQcsClient>,
48        timeout: Option<f64>,
49    ) -> PyResult<String> {
50        let client = PyQcsClient::get_or_create_client(client);
51        let timeout = timeout.map(Duration::from_secs_f64);
52        qcs::qpu::translation::get_quilt_calibrations(quantum_processor_id, &client, timeout)
53            .await.map_err(RustTranslationError::from).map_err(RustTranslationError::to_py_err)
54    }
55}
56
57#[derive(Debug, thiserror::Error)]
58pub enum RustTranslationError {
59    #[error(transparent)]
60    Translation(#[from] qcs::qpu::translation::Error),
61    #[error("Failed to serialize translation result: {0}")]
62    Serialization(#[from] serde_json::Error),
63}
64
65py_wrap_error!(
66    translation,
67    RustTranslationError,
68    TranslationError,
69    PyRuntimeError
70);
71
72#[derive(Copy, Clone)]
73pub enum TranslationBackend {
74    V1,
75    V2,
76}
77
78py_wrap_simple_enum! {
79    PyTranslationBackend(TranslationBackend) as "TranslationBackend" {
80        V1,
81        V2
82    }
83}
84
85#[derive(Clone, Default, Debug)]
86#[pyclass(name = "TranslationOptions")]
87pub struct PyTranslationOptions(TranslationOptions);
88
89impl PyTranslationOptions {
90    pub fn as_inner(&self) -> &TranslationOptions {
91        &self.0
92    }
93}
94
95#[pymethods]
96impl PyTranslationOptions {
97    #[new]
98    fn __new__() -> PyResult<Self> {
99        Ok(Self(Default::default()))
100    }
101
102    #[getter]
103    fn backend(&self) -> Option<PyTranslationBackend> {
104        self.0.backend().map(|b| match b {
105            ApiTranslationBackend::V1(_) => PyTranslationBackend::V1,
106            ApiTranslationBackend::V2(_) => PyTranslationBackend::V2,
107        })
108    }
109
110    fn use_backend_v1(&mut self) {
111        self.0.with_backend_v1();
112    }
113
114    fn use_backend_v2(&mut self) {
115        self.0.with_backend_v2();
116    }
117
118    #[pyo3(signature = (q_ctrl = PyQCtrl::default(), /))]
119    fn use_q_ctrl(&mut self, q_ctrl: PyQCtrl) {
120        self.0.q_ctrl(*q_ctrl.as_inner());
121    }
122
123    #[pyo3(signature = (riverlane = PyRiverlane::default(), /))]
124    fn use_riverlane(&mut self, riverlane: PyRiverlane) {
125        self.0.riverlane(riverlane.as_inner().clone());
126    }
127
128    #[staticmethod]
129    fn v1() -> Self {
130        let mut builder = TranslationOptions::default();
131        builder.with_backend_v1();
132        Self(builder)
133    }
134
135    #[staticmethod]
136    #[pyo3(signature = (
137        *,
138        prepend_default_calibrations=None,
139        passive_reset_delay_seconds=None,
140        allow_unchecked_pointer_arithmetic=None,
141        allow_frame_redefinition=None
142    ))]
143    fn v2(
144        prepend_default_calibrations: Option<bool>,
145        passive_reset_delay_seconds: Option<f64>,
146        allow_unchecked_pointer_arithmetic: Option<bool>,
147        allow_frame_redefinition: Option<bool>,
148    ) -> Self {
149        let mut builder = TranslationOptions::default();
150        builder.with_backend_v2();
151        if let Some(prepend) = prepend_default_calibrations {
152            builder
153                .v2_prepend_default_calibrations(prepend)
154                .expect("using the correct backend");
155        }
156        if let Some(delay) = passive_reset_delay_seconds {
157            builder
158                .v2_passive_reset_delay_seconds(delay)
159                .expect("using the correct backend");
160        }
161        if let Some(allow) = allow_unchecked_pointer_arithmetic {
162            builder
163                .v2_allow_unchecked_pointer_arithmetic(allow)
164                .expect("using the correct backend");
165        }
166        if let Some(allow) = allow_frame_redefinition {
167            builder
168                .v2_allow_frame_redefinition(allow)
169                .expect("using the correct backend");
170        }
171        Self(builder)
172    }
173
174    fn encode_as_protobuf<'a>(&'a self, py: Python<'a>) -> &'a PyBytes {
175        let options: ApiTranslationOptions = self.0.clone().into();
176        PyBytes::new(py, options.encode_to_vec().as_slice())
177    }
178
179    fn __repr__(&self) -> String {
180        format!("{:?}", self.0)
181    }
182}
183
184#[derive(Clone, Default, Debug)]
185#[pyclass(name = "QCtrl")]
186pub struct PyQCtrl(translation_options::QCtrl);
187
188#[pymethods]
189impl PyQCtrl {
190    #[new]
191    #[pyo3(signature = (/, fixed_layout = None))]
192    fn __new__(fixed_layout: Option<bool>) -> PyResult<Self> {
193        Ok(Self(translation_options::QCtrl { fixed_layout }))
194    }
195}
196
197impl PyQCtrl {
198    fn as_inner(&self) -> &translation_options::QCtrl {
199        &self.0
200    }
201}
202
203#[derive(Clone, Default, Debug)]
204#[pyclass(name = "Riverlane")]
205pub struct PyRiverlane(Riverlane);
206
207#[pymethods]
208impl PyRiverlane {
209    #[new]
210    #[pyo3(
211        signature = (qeci_configuration_data, qeci_max_nanoseconds_until_read_available, /)
212    )]
213    fn __new__(
214        qeci_configuration_data: HashMap<String, Vec<u8>>,
215        qeci_max_nanoseconds_until_read_available: u64,
216    ) -> PyResult<Self> {
217        Ok(Self(Riverlane {
218            qeci_configuration_data,
219            qeci_max_nanoseconds_until_read_available,
220        }))
221    }
222}
223
224impl PyRiverlane {
225    fn as_inner(&self) -> &Riverlane {
226        &self.0
227    }
228}
229
230/// The result of a call to [`translate`] which provides information about the
231/// translated program.
232#[pyclass]
233#[pyo3(name = "TranslationResult")]
234pub struct PyTranslationResult {
235    /// The translated program.
236    #[pyo3(get)]
237    pub program: String,
238
239    /// The memory locations used for readout.
240    #[pyo3(get)]
241    pub ro_sources: Option<HashMap<String, String>>,
242}
243
244crate::py_sync::py_function_sync_async! {
245    /// Translates a native Quil program into an executable
246    ///
247    /// # Errors
248    ///
249    /// Returns a [`TranslationError`] if translation fails.
250    #[pyfunction]
251    #[pyo3(signature = (native_quil, num_shots, quantum_processor_id, client = None, translation_options = None))]
252    #[pyo3_opentelemetry::pypropagate(on_context_extraction_failure="ignore")]
253    async fn translate(
254        native_quil: String,
255        num_shots: u32,
256        quantum_processor_id: String,
257        client: Option<PyQcsClient>,
258        translation_options: Option<PyTranslationOptions>,
259    ) -> PyResult<PyTranslationResult> {
260        let client = PyQcsClient::get_or_create_client(client);
261        let translation_options = translation_options.map(|opts| opts.as_inner().clone());
262        let result = qcs::qpu::translation::translate(&quantum_processor_id, &native_quil, num_shots, &client, translation_options).with_current_context()
263                .map_err(RustTranslationError::from)
264                .map_err(RustTranslationError::to_py_err).await?;
265
266        let program = serde_json::to_string(&result.job)
267            .map_err(RustTranslationError::from)
268            .map_err(RustTranslationError::to_py_err)?;
269
270        Ok(PyTranslationResult {
271            program,
272            ro_sources: Some(result.readout_map),
273        })
274    }
275}
qcs_sdk/qpu/translation.rs

qcs_sdk/qpu/
translation.rs
