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

spvirit/
server.rs
