spvirit/
server.rs

1//! Python server wrappers — sync-only for phase 1.
2
3use std::net::IpAddr;
4use std::sync::Arc;
5use std::time::Duration;
6
7use pyo3::prelude::*;
8
9use spvirit_codec::spvd_decode::DecodedValue;
10use spvirit_server::SimplePvStore;
11use spvirit_server::pva_server::PvaServer;
12use spvirit_types::{ScalarArrayValue, ScalarValue};
13
14use crate::convert::{decoded_to_py, py_to_scalar, py_to_scalar_array, scalar_to_py};
15use crate::nt::{nt_payload_to_py, py_to_nt_payload};
16use crate::runtime::RUNTIME;
17use crate::source::{PyNotifier, PySourceAdapter};
18
19// ─── ServerBuilder ───────────────────────────────────────────────────────────
20
21#[pyclass(name = "ServerBuilder")]
22pub struct PyServerBuilder {
23    builder: Option<spvirit_server::PvaServerBuilder>,
24    /// Python sources to wire up on build (label, order, adapter).
25    python_sources: Vec<(String, i32, Arc<PySourceAdapter>)>,
26}
27
28#[pymethods]
29impl PyServerBuilder {
30    #[new]
31    fn new() -> Self {
32        Self {
33            builder: Some(PvaServer::builder()),
34            python_sources: Vec::new(),
35        }
36    }
37
38    fn ai(mut slf: PyRefMut<'_, Self>, name: String, initial: f64) -> PyRefMut<'_, Self> {
39        let b = slf.builder.take().expect("builder consumed");
40        slf.builder = Some(b.ai(name, initial));
41        slf
42    }
43
44    fn ao(mut slf: PyRefMut<'_, Self>, name: String, initial: f64) -> PyRefMut<'_, Self> {
45        let b = slf.builder.take().expect("builder consumed");
46        slf.builder = Some(b.ao(name, initial));
47        slf
48    }
49
50    fn bi(mut slf: PyRefMut<'_, Self>, name: String, initial: bool) -> PyRefMut<'_, Self> {
51        let b = slf.builder.take().expect("builder consumed");
52        slf.builder = Some(b.bi(name, initial));
53        slf
54    }
55
56    fn bo(mut slf: PyRefMut<'_, Self>, name: String, initial: bool) -> PyRefMut<'_, Self> {
57        let b = slf.builder.take().expect("builder consumed");
58        slf.builder = Some(b.bo(name, initial));
59        slf
60    }
61
62    fn string_in(mut slf: PyRefMut<'_, Self>, name: String, initial: String) -> PyRefMut<'_, Self> {
63        let b = slf.builder.take().expect("builder consumed");
64        slf.builder = Some(b.string_in(name, initial));
65        slf
66    }
67
68    fn string_out(
69        mut slf: PyRefMut<'_, Self>,
70        name: String,
71        initial: String,
72    ) -> PyRefMut<'_, Self> {
73        let b = slf.builder.take().expect("builder consumed");
74        slf.builder = Some(b.string_out(name, initial));
75        slf
76    }
77
78    fn waveform<'py>(
79        mut slf: PyRefMut<'py, Self>,
80        name: String,
81        data: &Bound<'py, PyAny>,
82    ) -> PyResult<PyRefMut<'py, Self>> {
83        let arr = py_to_scalar_array(data)?;
84        let b = slf.builder.take().expect("builder consumed");
85        slf.builder = Some(b.waveform(name, arr));
86        Ok(slf)
87    }
88
89    fn aai<'py>(
90        mut slf: PyRefMut<'py, Self>,
91        name: String,
92        data: &Bound<'py, PyAny>,
93    ) -> PyResult<PyRefMut<'py, Self>> {
94        let arr = py_to_scalar_array(data)?;
95        let b = slf.builder.take().expect("builder consumed");
96        slf.builder = Some(b.aai(name, arr));
97        Ok(slf)
98    }
99
100    fn aao<'py>(
101        mut slf: PyRefMut<'py, Self>,
102        name: String,
103        data: &Bound<'py, PyAny>,
104    ) -> PyResult<PyRefMut<'py, Self>> {
105        let arr = py_to_scalar_array(data)?;
106        let b = slf.builder.take().expect("builder consumed");
107        slf.builder = Some(b.aao(name, arr));
108        Ok(slf)
109    }
110
111    #[pyo3(signature = (name, data, indx=0, nelm=None))]
112    fn sub_array<'py>(
113        mut slf: PyRefMut<'py, Self>,
114        name: String,
115        data: &Bound<'py, PyAny>,
116        indx: usize,
117        nelm: Option<usize>,
118    ) -> PyResult<PyRefMut<'py, Self>> {
119        let arr = py_to_scalar_array(data)?;
120        let n = nelm.unwrap_or(arr.len());
121        let b = slf.builder.take().expect("builder consumed");
122        slf.builder = Some(b.sub_array(name, arr, indx, n));
123        Ok(slf)
124    }
125
126    fn nt_table<'py>(
127        mut slf: PyRefMut<'py, Self>,
128        name: String,
129        columns: &Bound<'py, PyAny>,
130    ) -> PyResult<PyRefMut<'py, Self>> {
131        let dict = columns.downcast::<pyo3::types::PyDict>().map_err(|_| {
132            pyo3::exceptions::PyTypeError::new_err("columns must be a dict of {name: list}")
133        })?;
134        let mut cols: Vec<(String, ScalarArrayValue)> = Vec::new();
135        for (key, val) in dict.iter() {
136            let col_name: String = key.extract()?;
137            let col_data = py_to_scalar_array(&val)?;
138            cols.push((col_name, col_data));
139        }
140        let b = slf.builder.take().expect("builder consumed");
141        slf.builder = Some(b.nt_table(name, cols));
142        Ok(slf)
143    }
144
145    fn nt_ndarray<'py>(
146        mut slf: PyRefMut<'py, Self>,
147        name: String,
148        data: &Bound<'py, PyAny>,
149        dims: Vec<(i32, i32)>,
150    ) -> PyResult<PyRefMut<'py, Self>> {
151        let arr = py_to_scalar_array(data)?;
152        let b = slf.builder.take().expect("builder consumed");
153        slf.builder = Some(b.nt_ndarray(name, arr, dims));
154        Ok(slf)
155    }
156
157    fn mbbi(
158        mut slf: PyRefMut<'_, Self>,
159        name: String,
160        choices: Vec<String>,
161        initial: i32,
162    ) -> PyRefMut<'_, Self> {
163        let b = slf.builder.take().expect("builder consumed");
164        slf.builder = Some(b.mbbi(name, choices, initial));
165        slf
166    }
167
168    fn mbbo(
169        mut slf: PyRefMut<'_, Self>,
170        name: String,
171        choices: Vec<String>,
172        initial: i32,
173    ) -> PyRefMut<'_, Self> {
174        let b = slf.builder.take().expect("builder consumed");
175        slf.builder = Some(b.mbbo(name, choices, initial));
176        slf
177    }
178
179    fn generic<'py>(
180        mut slf: PyRefMut<'py, Self>,
181        name: String,
182        struct_id: String,
183        fields: &Bound<'py, pyo3::types::PyDict>,
184    ) -> PyResult<PyRefMut<'py, Self>> {
185        let mut field_vec: Vec<(String, spvirit_types::PvValue)> = Vec::new();
186        for (key, val) in fields.iter() {
187            let field_name: String = key.extract()?;
188            let pv_val = py_to_pv_value(&val)?;
189            field_vec.push((field_name, pv_val));
190        }
191        let b = slf.builder.take().expect("builder consumed");
192        slf.builder = Some(b.generic(name, struct_id, field_vec));
193        Ok(slf)
194    }
195
196    fn db_file(mut slf: PyRefMut<'_, Self>, path: String) -> PyRefMut<'_, Self> {
197        let b = slf.builder.take().expect("builder consumed");
198        slf.builder = Some(b.db_file(path));
199        slf
200    }
201
202    fn db_string(mut slf: PyRefMut<'_, Self>, content: String) -> PyRefMut<'_, Self> {
203        let b = slf.builder.take().expect("builder consumed");
204        slf.builder = Some(b.db_string(&content));
205        slf
206    }
207
208    fn on_put(mut slf: PyRefMut<'_, Self>, name: String, callback: PyObject) -> PyRefMut<'_, Self> {
209        let b = slf.builder.take().expect("builder consumed");
210        slf.builder = Some(
211            b.on_put(name, move |pv_name: &str, decoded: &DecodedValue| {
212                Python::with_gil(|py| {
213                    let py_val = decoded_to_py(py, decoded);
214                    if let Err(e) = callback.call1(py, (pv_name, py_val)) {
215                        tracing::error!("on_put callback error: {e}");
216                    }
217                });
218            }),
219        );
220        slf
221    }
222
223    fn scan(
224        mut slf: PyRefMut<'_, Self>,
225        name: String,
226        period_secs: f64,
227        callback: PyObject,
228    ) -> PyRefMut<'_, Self> {
229        let b = slf.builder.take().expect("builder consumed");
230        let dur = Duration::from_secs_f64(period_secs);
231        slf.builder = Some(b.scan(name, dur, move |pv_name: &str| {
232            Python::with_gil(|py| match callback.call1(py, (pv_name,)) {
233                Ok(ret) => py_to_scalar(ret.bind(py)).unwrap_or(ScalarValue::F64(0.0)),
234                Err(e) => {
235                    tracing::error!("scan callback error: {e}");
236                    ScalarValue::F64(0.0)
237                }
238            })
239        }));
240        slf
241    }
242
243    fn port(mut slf: PyRefMut<'_, Self>, port: u16) -> PyRefMut<'_, Self> {
244        let b = slf.builder.take().expect("builder consumed");
245        slf.builder = Some(b.port(port));
246        slf
247    }
248
249    fn udp_port(mut slf: PyRefMut<'_, Self>, port: u16) -> PyRefMut<'_, Self> {
250        let b = slf.builder.take().expect("builder consumed");
251        slf.builder = Some(b.udp_port(port));
252        slf
253    }
254
255    fn listen_ip(mut slf: PyRefMut<'_, Self>, ip: String) -> PyResult<PyRefMut<'_, Self>> {
256        let ip_addr: IpAddr = ip
257            .parse()
258            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
259        let b = slf.builder.take().expect("builder consumed");
260        slf.builder = Some(b.listen_ip(ip_addr));
261        Ok(slf)
262    }
263
264    fn advertise_ip(mut slf: PyRefMut<'_, Self>, ip: String) -> PyResult<PyRefMut<'_, Self>> {
265        let ip_addr: IpAddr = ip
266            .parse()
267            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
268        let b = slf.builder.take().expect("builder consumed");
269        slf.builder = Some(b.advertise_ip(ip_addr));
270        Ok(slf)
271    }
272
273    fn compute_alarms(mut slf: PyRefMut<'_, Self>, enabled: bool) -> PyRefMut<'_, Self> {
274        let b = slf.builder.take().expect("builder consumed");
275        slf.builder = Some(b.compute_alarms(enabled));
276        slf
277    }
278
279    fn beacon_period(mut slf: PyRefMut<'_, Self>, secs: u64) -> PyRefMut<'_, Self> {
280        let b = slf.builder.take().expect("builder consumed");
281        slf.builder = Some(b.beacon_period(secs));
282        slf
283    }
284
285    /// Register a Python-defined [`Source`].
286    ///
287    /// `source` is any Python object implementing `claim`, `get`, `put`,
288    /// `names`, and (optionally) `rpc` / `on_start`.  See the
289    /// `demo_source_*.py` examples for patterns.
290    ///
291    /// Lower `order` values are tried first during PV name resolution;
292    /// the built-in record store is always at order 0.
293    fn add_source(
294        mut slf: PyRefMut<'_, Self>,
295        label: String,
296        order: i32,
297        source: PyObject,
298    ) -> PyRefMut<'_, Self> {
299        let adapter = Arc::new(PySourceAdapter::new(source));
300        slf.python_sources
301            .push((label.clone(), order, adapter.clone()));
302        let b = slf.builder.take().expect("builder consumed");
303        // Cast to Arc<dyn Source> via Arc<PySourceAdapter>.
304        let as_dyn: Arc<dyn spvirit_server::pvstore::Source> = adapter;
305        slf.builder = Some(b.source(label, order, as_dyn));
306        slf
307    }
308
309    /// Build and return a `Server` that can be started.
310    fn build(&mut self) -> PyResult<PyServer> {
311        let b = self
312            .builder
313            .take()
314            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("builder already consumed"))?;
315        let mut server = b.build();
316        let store = server.store().clone();
317        // Pre-create the monitor registry so Python sources can notify
318        // PVAccess monitor subscribers before .run() starts.
319        let registry = server.monitor_registry();
320        let notifier = PyNotifier::new(registry);
321        let sources = std::mem::take(&mut self.python_sources);
322        // Invoke `on_start(notifier)` on every Python source that defines it.
323        for (_, _, adapter) in &sources {
324            adapter.invoke_on_start(notifier.clone());
325        }
326        Ok(PyServer {
327            server: Some(server),
328            store: Some(store),
329            notifier: Some(notifier),
330            post_build_sources: sources,
331        })
332    }
333}
334
335// ─── Server ──────────────────────────────────────────────────────────────────
336
337#[pyclass(name = "Server")]
338pub struct PyServer {
339    server: Option<PvaServer>,
340    store: Option<Arc<SimplePvStore>>,
341    /// Notifier handed to each Python source so it can publish monitor updates.
342    notifier: Option<PyNotifier>,
343    /// Adapters for all Python sources registered on this server — kept alive
344    /// so they outlive `run()`.
345    #[allow(dead_code)]
346    post_build_sources: Vec<(String, i32, Arc<PySourceAdapter>)>,
347}
348
349#[pymethods]
350impl PyServer {
351    /// Get a handle to the PV store for runtime get/set.
352    fn store(&self) -> PyResult<PyStore> {
353        let store = self
354            .store
355            .as_ref()
356            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?
357            .clone();
358        Ok(PyStore { inner: store })
359    }
360
361    /// Return the monitor notifier for publishing updates from Python code.
362    fn notifier(&self) -> PyResult<PyNotifier> {
363        self.notifier
364            .clone()
365            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))
366    }
367
368    /// Register an additional Python source after build.  The source's
369    /// `on_start(notifier)` (if defined) is invoked immediately.
370    fn add_source(&mut self, label: String, order: i32, source: PyObject) -> PyResult<()> {
371        let server = self
372            .server
373            .as_mut()
374            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
375        let adapter = Arc::new(PySourceAdapter::new(source));
376        if let Some(notifier) = self.notifier.clone() {
377            adapter.invoke_on_start(notifier);
378        }
379        let as_dyn: Arc<dyn spvirit_server::pvstore::Source> = adapter.clone();
380        server.add_source(label.clone(), order, as_dyn);
381        self.post_build_sources.push((label, order, adapter));
382        Ok(())
383    }
384
385    /// Run the server (blocking). This does not return until the server stops.
386    fn run(&mut self, py: Python<'_>) -> PyResult<()> {
387        let server = self
388            .server
389            .take()
390            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
391        py.allow_threads(|| {
392            RUNTIME
393                .block_on(server.run())
394                .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
395        })
396    }
397
398    /// Start the server in a background thread and return the store handle.
399    fn start_background(&mut self) -> PyResult<PyStore> {
400        let server = self
401            .server
402            .take()
403            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
404        let store = self
405            .store
406            .as_ref()
407            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?
408            .clone();
409
410        std::thread::spawn(move || {
411            if let Err(e) = RUNTIME.block_on(server.run()) {
412                tracing::error!("background server error: {e}");
413            }
414        });
415
416        Ok(PyStore { inner: store })
417    }
418}
419
420// ─── Store ───────────────────────────────────────────────────────────────────
421
422#[pyclass(name = "Store")]
423pub struct PyStore {
424    inner: Arc<SimplePvStore>,
425}
426
427#[pymethods]
428impl PyStore {
429    /// Get the current scalar value of a PV (returns None if not found).
430    fn get_value(&self, py: Python<'_>, name: String) -> PyResult<PyObject> {
431        let store = self.inner.clone();
432        let val = py.allow_threads(|| RUNTIME.block_on(store.get_value(&name)));
433        Ok(match val {
434            Some(v) => scalar_to_py(py, &v),
435            None => py.None(),
436        })
437    }
438
439    /// Get the full NT payload for a PV (returns NtScalar, NtScalarArray, etc.).
440    fn get_nt(&self, py: Python<'_>, name: String) -> PyResult<PyObject> {
441        let store = self.inner.clone();
442        let val = py.allow_threads(|| RUNTIME.block_on(store.get_nt(&name)));
443        Ok(match val {
444            Some(payload) => nt_payload_to_py(py, payload),
445            None => py.None(),
446        })
447    }
448
449    /// Set a scalar value on a PV. Returns True if the PV exists.
450    fn set_value(&self, py: Python<'_>, name: String, value: &Bound<'_, PyAny>) -> PyResult<bool> {
451        let sv = py_to_scalar(value)?;
452        let store = self.inner.clone();
453        Ok(py.allow_threads(|| RUNTIME.block_on(store.set_value(&name, sv))))
454    }
455
456    /// Set an array value on a PV. Returns True if the PV exists.
457    fn set_array_value(
458        &self,
459        py: Python<'_>,
460        name: String,
461        value: &Bound<'_, PyAny>,
462    ) -> PyResult<bool> {
463        let arr = py_to_scalar_array(value)?;
464        let store = self.inner.clone();
465        Ok(py.allow_threads(|| RUNTIME.block_on(store.set_array_value(&name, arr))))
466    }
467
468    /// Write a full NT payload (NtScalar, NtScalarArray, etc.) to a PV.
469    /// Returns True if the PV exists.
470    fn put_nt(&self, py: Python<'_>, name: String, nt: &Bound<'_, PyAny>) -> PyResult<bool> {
471        let payload = py_to_nt_payload(nt)?;
472        let store = self.inner.clone();
473        Ok(py.allow_threads(|| RUNTIME.block_on(store.put_nt(&name, payload))))
474    }
475
476    /// List all PV names in the store.
477    fn pv_names(&self, py: Python<'_>) -> PyResult<Vec<String>> {
478        let store = self.inner.clone();
479        Ok(py.allow_threads(|| RUNTIME.block_on(store.pv_names())))
480    }
481}
482
483// ─── Helpers ─────────────────────────────────────────────────────────────────
484
485/// Convert a Python value to a [`PvValue`].
486///
487/// Scalars (bool, int, float, str) become `PvValue::Scalar`.
488/// Lists become `PvValue::ScalarArray`.
489fn py_to_pv_value(obj: &Bound<'_, PyAny>) -> PyResult<spvirit_types::PvValue> {
490    if let Ok(list) = obj.downcast::<pyo3::types::PyList>() {
491        let arr = py_to_scalar_array(list.as_any())?;
492        Ok(spvirit_types::PvValue::ScalarArray(arr))
493    } else {
494        let sv = py_to_scalar(obj)?;
495        Ok(spvirit_types::PvValue::Scalar(sv))
496    }
497}
spvirit/server.rs

spvirit/
server.rs
