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, block_on_py};
17use crate::source::{PyNotifier, PySourceAdapter};
18
19// ─── ServerBuilder ───────────────────────────────────────────────────────────
20
21/// Fluent builder for a PVAccess server. Chain record definitions and
22/// configuration, then call `build()`. Single-use: any method called after
23/// `build()` raises RuntimeError.
24#[pyclass(name = "ServerBuilder")]
25pub struct PyServerBuilder {
26    builder: Option<spvirit_server::PvaServerBuilder>,
27    /// Python sources to wire up on build (label, order, adapter).
28    python_sources: Vec<(String, i32, Arc<PySourceAdapter>)>,
29}
30
31/// Take the inner builder, raising `RuntimeError` if `build()` already ran.
32fn take_builder(
33    slf: &mut PyRefMut<'_, PyServerBuilder>,
34) -> PyResult<spvirit_server::PvaServerBuilder> {
35    slf.builder.take().ok_or_else(|| {
36        pyo3::exceptions::PyRuntimeError::new_err(
37            "ServerBuilder already consumed by build(); create a new builder",
38        )
39    })
40}
41
42#[pymethods]
43impl PyServerBuilder {
44    #[new]
45    fn new() -> Self {
46        Self {
47            builder: Some(PvaServer::builder()),
48            python_sources: Vec::new(),
49        }
50    }
51
52    /// Add an `ai` (analog input) NTScalar double record — read-only over the wire.
53    fn ai(mut slf: PyRefMut<'_, Self>, name: String, initial: f64) -> PyResult<PyRefMut<'_, Self>> {
54        let b = take_builder(&mut slf)?;
55        slf.builder = Some(b.ai(name, initial));
56        Ok(slf)
57    }
58
59    /// Add an `ao` (analog output) NTScalar double record — writable over the wire.
60    fn ao(mut slf: PyRefMut<'_, Self>, name: String, initial: f64) -> PyResult<PyRefMut<'_, Self>> {
61        let b = take_builder(&mut slf)?;
62        slf.builder = Some(b.ao(name, initial));
63        Ok(slf)
64    }
65
66    /// Add a `bi` (binary input) NTScalar boolean record — read-only over the wire.
67    fn bi(
68        mut slf: PyRefMut<'_, Self>,
69        name: String,
70        initial: bool,
71    ) -> PyResult<PyRefMut<'_, Self>> {
72        let b = take_builder(&mut slf)?;
73        slf.builder = Some(b.bi(name, initial));
74        Ok(slf)
75    }
76
77    /// Add a `bo` (binary output) NTScalar boolean record — writable over the wire.
78    fn bo(
79        mut slf: PyRefMut<'_, Self>,
80        name: String,
81        initial: bool,
82    ) -> PyResult<PyRefMut<'_, Self>> {
83        let b = take_builder(&mut slf)?;
84        slf.builder = Some(b.bo(name, initial));
85        Ok(slf)
86    }
87
88    /// Add a `stringin` NTScalar string record — read-only over the wire.
89    fn string_in(
90        mut slf: PyRefMut<'_, Self>,
91        name: String,
92        initial: String,
93    ) -> PyResult<PyRefMut<'_, Self>> {
94        let b = take_builder(&mut slf)?;
95        slf.builder = Some(b.string_in(name, initial));
96        Ok(slf)
97    }
98
99    /// Add a `stringout` NTScalar string record — writable over the wire.
100    fn string_out(
101        mut slf: PyRefMut<'_, Self>,
102        name: String,
103        initial: String,
104    ) -> PyResult<PyRefMut<'_, Self>> {
105        let b = take_builder(&mut slf)?;
106        slf.builder = Some(b.string_out(name, initial));
107        Ok(slf)
108    }
109
110    /// Add a `waveform` NTScalarArray record — writable over the wire.
111    fn waveform<'py>(
112        mut slf: PyRefMut<'py, Self>,
113        name: String,
114        data: &Bound<'py, PyAny>,
115    ) -> PyResult<PyRefMut<'py, Self>> {
116        let arr = py_to_scalar_array(data)?;
117        let b = take_builder(&mut slf)?;
118        slf.builder = Some(b.waveform(name, arr));
119        Ok(slf)
120    }
121
122    /// Add an `aai` (analog array input) NTScalarArray record — read-only over the wire.
123    fn aai<'py>(
124        mut slf: PyRefMut<'py, Self>,
125        name: String,
126        data: &Bound<'py, PyAny>,
127    ) -> PyResult<PyRefMut<'py, Self>> {
128        let arr = py_to_scalar_array(data)?;
129        let b = take_builder(&mut slf)?;
130        slf.builder = Some(b.aai(name, arr));
131        Ok(slf)
132    }
133
134    /// Add an `aao` (analog array output) NTScalarArray record — writable over the wire.
135    fn aao<'py>(
136        mut slf: PyRefMut<'py, Self>,
137        name: String,
138        data: &Bound<'py, PyAny>,
139    ) -> PyResult<PyRefMut<'py, Self>> {
140        let arr = py_to_scalar_array(data)?;
141        let b = take_builder(&mut slf)?;
142        slf.builder = Some(b.aao(name, arr));
143        Ok(slf)
144    }
145
146    /// Add a `subArray` record serving a `nelm`-element window of `data`
147    /// starting at `indx` (defaults to the full array).
148    #[pyo3(signature = (name, data, indx=0, nelm=None))]
149    fn sub_array<'py>(
150        mut slf: PyRefMut<'py, Self>,
151        name: String,
152        data: &Bound<'py, PyAny>,
153        indx: usize,
154        nelm: Option<usize>,
155    ) -> PyResult<PyRefMut<'py, Self>> {
156        let arr = py_to_scalar_array(data)?;
157        let n = nelm.unwrap_or(arr.len());
158        let b = take_builder(&mut slf)?;
159        slf.builder = Some(b.sub_array(name, arr, indx, n));
160        Ok(slf)
161    }
162
163    /// Add an NTTable record from a `{column_name: list}` dict of columns.
164    fn nt_table<'py>(
165        mut slf: PyRefMut<'py, Self>,
166        name: String,
167        columns: &Bound<'py, PyAny>,
168    ) -> PyResult<PyRefMut<'py, Self>> {
169        let dict = columns.downcast::<pyo3::types::PyDict>().map_err(|_| {
170            pyo3::exceptions::PyTypeError::new_err("columns must be a dict of {name: list}")
171        })?;
172        let mut cols: Vec<(String, ScalarArrayValue)> = Vec::new();
173        for (key, val) in dict.iter() {
174            let col_name: String = key.extract()?;
175            let col_data = py_to_scalar_array(&val)?;
176            cols.push((col_name, col_data));
177        }
178        let b = take_builder(&mut slf)?;
179        slf.builder = Some(b.nt_table(name, cols));
180        Ok(slf)
181    }
182
183    /// Add an NTNDArray record from flat array data and `(size, full_size)`
184    /// dimension pairs.
185    fn nt_ndarray<'py>(
186        mut slf: PyRefMut<'py, Self>,
187        name: String,
188        data: &Bound<'py, PyAny>,
189        dims: Vec<(i32, i32)>,
190    ) -> PyResult<PyRefMut<'py, Self>> {
191        let arr = py_to_scalar_array(data)?;
192        let b = take_builder(&mut slf)?;
193        slf.builder = Some(b.nt_ndarray(name, arr, dims));
194        Ok(slf)
195    }
196
197    /// Add an `mbbi` (multi-bit binary input) NTEnum record — read-only over
198    /// the wire. `initial` is the choice index.
199    fn mbbi(
200        mut slf: PyRefMut<'_, Self>,
201        name: String,
202        choices: Vec<String>,
203        initial: i32,
204    ) -> PyResult<PyRefMut<'_, Self>> {
205        let b = take_builder(&mut slf)?;
206        slf.builder = Some(b.mbbi(name, choices, initial));
207        Ok(slf)
208    }
209
210    /// Add an `mbbo` (multi-bit binary output) NTEnum record — writable over
211    /// the wire. `initial` is the choice index.
212    fn mbbo(
213        mut slf: PyRefMut<'_, Self>,
214        name: String,
215        choices: Vec<String>,
216        initial: i32,
217    ) -> PyResult<PyRefMut<'_, Self>> {
218        let b = take_builder(&mut slf)?;
219        slf.builder = Some(b.mbbo(name, choices, initial));
220        Ok(slf)
221    }
222
223    /// Add a generic structure record with the given struct ID and a
224    /// `{field_name: value}` dict (scalars or lists).
225    fn generic<'py>(
226        mut slf: PyRefMut<'py, Self>,
227        name: String,
228        struct_id: String,
229        fields: &Bound<'py, pyo3::types::PyDict>,
230    ) -> PyResult<PyRefMut<'py, Self>> {
231        let mut field_vec: Vec<(String, spvirit_types::PvValue)> = Vec::new();
232        for (key, val) in fields.iter() {
233            let field_name: String = key.extract()?;
234            let pv_val = py_to_pv_value(&val)?;
235            field_vec.push((field_name, pv_val));
236        }
237        let b = take_builder(&mut slf)?;
238        slf.builder = Some(b.generic(name, struct_id, field_vec));
239        Ok(slf)
240    }
241
242    /// Load record definitions from an EPICS `.db` file at `path`.
243    fn db_file(mut slf: PyRefMut<'_, Self>, path: String) -> PyResult<PyRefMut<'_, Self>> {
244        let b = take_builder(&mut slf)?;
245        slf.builder = Some(b.db_file(path));
246        Ok(slf)
247    }
248
249    /// Load record definitions from EPICS `.db` text given as a string.
250    fn db_string(mut slf: PyRefMut<'_, Self>, content: String) -> PyResult<PyRefMut<'_, Self>> {
251        let b = take_builder(&mut slf)?;
252        slf.builder = Some(b.db_string(&content));
253        Ok(slf)
254    }
255
256    fn on_put(
257        mut slf: PyRefMut<'_, Self>,
258        name: String,
259        callback: PyObject,
260    ) -> PyResult<PyRefMut<'_, Self>> {
261        let b = take_builder(&mut slf)?;
262        slf.builder = Some(
263            b.on_put(name, move |pv_name: &str, decoded: &DecodedValue| {
264                Python::with_gil(|py| {
265                    let py_val = decoded_to_py(py, decoded);
266                    if let Err(e) = callback.call1(py, (pv_name, py_val)) {
267                        tracing::error!("on_put callback error: {e}");
268                    }
269                });
270            }),
271        );
272        Ok(slf)
273    }
274
275    fn scan(
276        mut slf: PyRefMut<'_, Self>,
277        name: String,
278        period: f64,
279        callback: PyObject,
280    ) -> PyResult<PyRefMut<'_, Self>> {
281        let b = take_builder(&mut slf)?;
282        let dur = Duration::from_secs_f64(period);
283        slf.builder = Some(b.scan(name, dur, move |pv_name: &str| {
284            Python::with_gil(|py| match callback.call1(py, (pv_name,)) {
285                Ok(ret) => py_to_scalar(ret.bind(py)).unwrap_or(ScalarValue::F64(0.0)),
286                Err(e) => {
287                    tracing::error!("scan callback error: {e}");
288                    ScalarValue::F64(0.0)
289                }
290            })
291        }));
292        Ok(slf)
293    }
294
295    /// Set the TCP port the server listens on.
296    fn port(mut slf: PyRefMut<'_, Self>, port: u16) -> PyResult<PyRefMut<'_, Self>> {
297        let b = take_builder(&mut slf)?;
298        slf.builder = Some(b.port(port));
299        Ok(slf)
300    }
301
302    /// Set the UDP port used for search requests and beacons.
303    fn udp_port(mut slf: PyRefMut<'_, Self>, port: u16) -> PyResult<PyRefMut<'_, Self>> {
304        let b = take_builder(&mut slf)?;
305        slf.builder = Some(b.udp_port(port));
306        Ok(slf)
307    }
308
309    /// Set the IP address to bind listeners to. Raises ValueError on an
310    /// invalid IP string.
311    fn listen_ip(mut slf: PyRefMut<'_, Self>, ip: String) -> PyResult<PyRefMut<'_, Self>> {
312        let ip_addr: IpAddr = ip
313            .parse()
314            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
315        let b = take_builder(&mut slf)?;
316        slf.builder = Some(b.listen_ip(ip_addr));
317        Ok(slf)
318    }
319
320    /// Set the IP address advertised to clients in search responses and
321    /// beacons. Raises ValueError on an invalid IP string.
322    fn advertise_ip(mut slf: PyRefMut<'_, Self>, ip: String) -> PyResult<PyRefMut<'_, Self>> {
323        let ip_addr: IpAddr = ip
324            .parse()
325            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
326        let b = take_builder(&mut slf)?;
327        slf.builder = Some(b.advertise_ip(ip_addr));
328        Ok(slf)
329    }
330
331    /// Enable or disable automatic alarm computation from record limits.
332    fn compute_alarms(mut slf: PyRefMut<'_, Self>, enabled: bool) -> PyResult<PyRefMut<'_, Self>> {
333        let b = take_builder(&mut slf)?;
334        slf.builder = Some(b.compute_alarms(enabled));
335        Ok(slf)
336    }
337
338    /// Set the UDP beacon period in seconds (float, rounded to whole
339    /// seconds, minimum 1).
340    fn beacon_period(mut slf: PyRefMut<'_, Self>, secs: f64) -> PyResult<PyRefMut<'_, Self>> {
341        let b = take_builder(&mut slf)?;
342        slf.builder = Some(b.beacon_period(secs.round().max(1.0) as u64));
343        Ok(slf)
344    }
345
346    fn __repr__(&self) -> &'static str {
347        if self.builder.is_some() {
348            "<spvirit.ServerBuilder>"
349        } else {
350            "<spvirit.ServerBuilder (consumed)>"
351        }
352    }
353
354    /// Register a Python-defined [`Source`].
355    ///
356    /// `source` is any Python object implementing `claim`, `get`, `put`,
357    /// `names`, and (optionally) `rpc` / `on_start`.  See the
358    /// `demo_source_*.py` examples for patterns.
359    ///
360    /// Lower `order` values are tried first during PV name resolution;
361    /// the built-in record store is always at order 0.
362    fn add_source(
363        mut slf: PyRefMut<'_, Self>,
364        label: String,
365        order: i32,
366        source: PyObject,
367    ) -> PyResult<PyRefMut<'_, Self>> {
368        let adapter = Arc::new(PySourceAdapter::new(source));
369        slf.python_sources
370            .push((label.clone(), order, adapter.clone()));
371        let b = take_builder(&mut slf)?;
372        // Cast to Arc<dyn Source> via Arc<PySourceAdapter>.
373        let as_dyn: Arc<dyn spvirit_server::pvstore::Source> = adapter;
374        slf.builder = Some(b.source(label, order, as_dyn));
375        Ok(slf)
376    }
377
378    /// Build and return a `Server` that can be started.
379    fn build(&mut self) -> PyResult<PyServer> {
380        let b = self
381            .builder
382            .take()
383            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("builder already consumed"))?;
384        let mut server = b.build();
385        let store = server.store().clone();
386        // Pre-create the monitor registry so Python sources can notify
387        // PVAccess monitor subscribers before .run() starts.
388        let registry = server.monitor_registry();
389        let notifier = PyNotifier::new(registry);
390        let sources = std::mem::take(&mut self.python_sources);
391        // Invoke `on_start(notifier)` on every Python source that defines it.
392        for (_, _, adapter) in &sources {
393            adapter.invoke_on_start(notifier.clone());
394        }
395        Ok(PyServer {
396            server: Some(server),
397            store: Some(store),
398            notifier: Some(notifier),
399            post_build_sources: sources,
400        })
401    }
402}
403
404// ─── Server ──────────────────────────────────────────────────────────────────
405
406/// A PVAccess server. Construct with `Server(pvs=..., ...)` or
407/// `ServerBuilder.build()`; start with `start()`, `run()`, or
408/// `start_background()`.
409#[pyclass(name = "Server")]
410pub struct PyServer {
411    server: Option<PvaServer>,
412    store: Option<Arc<SimplePvStore>>,
413    /// Notifier handed to each Python source so it can publish monitor updates.
414    notifier: Option<PyNotifier>,
415    /// Adapters for all Python sources registered on this server — kept alive
416    /// so they outlive `run()`.
417    #[allow(dead_code)]
418    post_build_sources: Vec<(String, i32, Arc<PySourceAdapter>)>,
419}
420
421#[pymethods]
422impl PyServer {
423    /// Build a server from typed PV handles (`spvirit.ai(...)` etc.).
424    ///
425    /// `pvs` — list of `Pv` handles; `sources` — list of `(label, order, obj)`
426    /// tuples of Python `Source` objects; remaining kwargs mirror
427    /// `ServerBuilder` configuration.
428    #[new]
429    #[pyo3(signature = (*, pvs=None, db_file=None, db_string=None, sources=None,
430                        port=None, udp_port=None, listen_ip=None, advertise_ip=None,
431                        compute_alarms=None, beacon_period=None))]
432    #[allow(clippy::too_many_arguments)]
433    fn new(
434        py: Python<'_>,
435        pvs: Option<Vec<crate::pv::PyPv>>,
436        db_file: Option<String>,
437        db_string: Option<String>,
438        sources: Option<Vec<(String, i32, PyObject)>>,
439        port: Option<u16>,
440        udp_port: Option<u16>,
441        listen_ip: Option<String>,
442        advertise_ip: Option<String>,
443        compute_alarms: Option<bool>,
444        beacon_period: Option<f64>,
445    ) -> PyResult<Self> {
446        let handles: Vec<spvirit_server::pv::AnyPv> =
447            pvs.unwrap_or_default().iter().map(|p| p.any()).collect();
448        let mut sb = PvaServer::serve(handles);
449        if let Some(p) = db_file {
450            sb = sb.db_file(p);
451        }
452        if let Some(s) = db_string {
453            sb = sb.db_string(&s);
454        }
455        if let Some(p) = port {
456            sb = sb.port(p);
457        }
458        if let Some(p) = udp_port {
459            sb = sb.udp_port(p);
460        }
461        if let Some(ip) = listen_ip {
462            let addr: IpAddr = ip
463                .parse()
464                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
465            sb = sb.listen_ip(addr);
466        }
467        if let Some(ip) = advertise_ip {
468            let addr: IpAddr = ip
469                .parse()
470                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid IP: {e}")))?;
471            sb = sb.advertise_ip(addr);
472        }
473        if let Some(c) = compute_alarms {
474            sb = sb.compute_alarms(c);
475        }
476        if let Some(secs) = beacon_period {
477            sb = sb.beacon_period(secs.round().max(1.0) as u64);
478        }
479        let mut python_sources: Vec<(String, i32, Arc<PySourceAdapter>)> = Vec::new();
480        for (label, order, obj) in sources.unwrap_or_default() {
481            let adapter = Arc::new(PySourceAdapter::new(obj));
482            python_sources.push((label.clone(), order, adapter.clone()));
483            let as_dyn: Arc<dyn spvirit_server::pvstore::Source> = adapter;
484            sb = sb.source(label, order, as_dyn);
485        }
486        let mut server = py.allow_threads(|| RUNTIME.block_on(sb.build()));
487        let store = server.store().clone();
488        let registry = server.monitor_registry();
489        let notifier = PyNotifier::new(registry);
490        for (_, _, adapter) in &python_sources {
491            adapter.invoke_on_start(notifier.clone());
492        }
493        Ok(PyServer {
494            server: Some(server),
495            store: Some(store),
496            notifier: Some(notifier),
497            post_build_sources: python_sources,
498        })
499    }
500
501    /// Return a fresh `ServerBuilder` (equivalent to `ServerBuilder()`).
502    #[staticmethod]
503    fn builder() -> PyServerBuilder {
504        PyServerBuilder::new()
505    }
506
507    /// Start serving on a background thread (returns immediately).
508    fn start(&mut self) -> PyResult<()> {
509        self.start_background().map(|_| ())
510    }
511
512    fn __repr__(&self) -> &'static str {
513        if self.server.is_some() {
514            "<spvirit.Server>"
515        } else {
516            "<spvirit.Server (running)>"
517        }
518    }
519
520    /// Mint a typed handle to any served record (handle-built or .db-loaded).
521    fn pv(&self, py: Python<'_>, name: String) -> PyResult<crate::pv::PyPv> {
522        use crate::pv::{PvKind, PyPv, pv_err};
523        use spvirit_types::{NtPayload, ScalarValue};
524        let server = self
525            .server
526            .as_ref()
527            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
528        let store = server.store().clone();
529        let sniff = block_on_py(py, store.get_nt(&name));
530        let kind = match sniff {
531            None => {
532                return Err(pyo3::exceptions::PyKeyError::new_err(format!(
533                    "PV '{name}' not found"
534                )));
535            }
536            Some(NtPayload::Scalar(nt)) => match nt.value {
537                ScalarValue::F64(_) | ScalarValue::F32(_) => {
538                    let h = block_on_py(py, server.pv::<f64>(&name)).map_err(pv_err)?;
539                    PvKind::F64(h)
540                }
541                ScalarValue::Bool(_) => {
542                    let h = block_on_py(py, server.pv::<bool>(&name)).map_err(pv_err)?;
543                    PvKind::Bool(h)
544                }
545                ScalarValue::I8(_) | ScalarValue::I16(_) | ScalarValue::I32(_) => {
546                    let h = block_on_py(py, server.pv::<i32>(&name)).map_err(pv_err)?;
547                    PvKind::I32(h)
548                }
549                ScalarValue::Str(_) => {
550                    let h = block_on_py(py, server.pv::<String>(&name)).map_err(pv_err)?;
551                    PvKind::Str(h)
552                }
553                other => {
554                    return Err(pyo3::exceptions::PyKeyError::new_err(format!(
555                        "PV '{name}' has unsupported value type {other:?} for typed handles"
556                    )));
557                }
558            },
559            Some(NtPayload::Enum(_)) => {
560                let h = block_on_py(py, server.pv::<i32>(&name)).map_err(pv_err)?;
561                PvKind::I32(h)
562            }
563            Some(NtPayload::ScalarArray(_)) => {
564                let h = block_on_py(py, server.array_pv(&name)).map_err(pv_err)?;
565                PvKind::Array(h)
566            }
567            Some(other) => {
568                return Err(pyo3::exceptions::PyKeyError::new_err(format!(
569                    "PV '{name}' has unsupported payload {other:?} for typed handles"
570                )));
571            }
572        };
573        Ok(PyPv { kind })
574    }
575
576    /// Get a handle to the PV store for runtime get/set.
577    fn store(&self) -> PyResult<PyStore> {
578        let store = self
579            .store
580            .as_ref()
581            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?
582            .clone();
583        Ok(PyStore { inner: store })
584    }
585
586    /// Return the monitor notifier for publishing updates from Python code.
587    fn notifier(&self) -> PyResult<PyNotifier> {
588        self.notifier
589            .clone()
590            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))
591    }
592
593    /// Register an additional Python source after build.  The source's
594    /// `on_start(notifier)` (if defined) is invoked immediately.
595    fn add_source(&mut self, label: String, order: i32, source: PyObject) -> PyResult<()> {
596        let server = self
597            .server
598            .as_mut()
599            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
600        let adapter = Arc::new(PySourceAdapter::new(source));
601        if let Some(notifier) = self.notifier.clone() {
602            adapter.invoke_on_start(notifier);
603        }
604        let as_dyn: Arc<dyn spvirit_server::pvstore::Source> = adapter.clone();
605        server.add_source(label.clone(), order, as_dyn);
606        self.post_build_sources.push((label, order, adapter));
607        Ok(())
608    }
609
610    /// Run the server (blocking). This does not return until the server stops.
611    fn run(&mut self, py: Python<'_>) -> PyResult<()> {
612        let server = self
613            .server
614            .take()
615            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
616        py.allow_threads(|| {
617            RUNTIME
618                .block_on(server.run())
619                .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
620        })
621    }
622
623    /// Start the server in a background thread and return the store handle.
624    fn start_background(&mut self) -> PyResult<PyStore> {
625        let server = self
626            .server
627            .take()
628            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?;
629        let store = self
630            .store
631            .as_ref()
632            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("server already consumed"))?
633            .clone();
634
635        std::thread::spawn(move || {
636            if let Err(e) = RUNTIME.block_on(server.run()) {
637                tracing::error!("background server error: {e}");
638            }
639        });
640
641        Ok(PyStore { inner: store })
642    }
643}
644
645// ─── Store ───────────────────────────────────────────────────────────────────
646
647/// Name-keyed runtime access to the server's record store: get/set scalar,
648/// array, and full NT values.
649#[pyclass(name = "Store")]
650pub struct PyStore {
651    inner: Arc<SimplePvStore>,
652}
653
654#[pymethods]
655impl PyStore {
656    /// Get the current scalar value of a PV (returns None if not found).
657    fn get_value(&self, py: Python<'_>, name: String) -> PyResult<PyObject> {
658        let store = self.inner.clone();
659        let val = block_on_py(py, store.get_value(&name));
660        Ok(match val {
661            Some(v) => scalar_to_py(py, &v),
662            None => py.None(),
663        })
664    }
665
666    /// Get the full NT payload for a PV (returns NtScalar, NtScalarArray, etc.).
667    fn get_nt(&self, py: Python<'_>, name: String) -> PyResult<PyObject> {
668        let store = self.inner.clone();
669        let val = block_on_py(py, store.get_nt(&name));
670        Ok(match val {
671            Some(payload) => nt_payload_to_py(py, payload),
672            None => py.None(),
673        })
674    }
675
676    /// Set a scalar value on a PV. Returns True if the PV exists.
677    fn set_value(&self, py: Python<'_>, name: String, value: &Bound<'_, PyAny>) -> PyResult<bool> {
678        let sv = py_to_scalar(value)?;
679        let store = self.inner.clone();
680        Ok(block_on_py(py, store.set_value(&name, sv)))
681    }
682
683    /// Set an array value on a PV. Returns True if the PV exists.
684    fn set_array_value(
685        &self,
686        py: Python<'_>,
687        name: String,
688        value: &Bound<'_, PyAny>,
689    ) -> PyResult<bool> {
690        let arr = py_to_scalar_array(value)?;
691        let store = self.inner.clone();
692        Ok(block_on_py(py, store.set_array_value(&name, arr)))
693    }
694
695    /// Write a full NT payload (NtScalar, NtScalarArray, etc.) to a PV.
696    /// Returns True if the PV exists.
697    fn put_nt(&self, py: Python<'_>, name: String, nt: &Bound<'_, PyAny>) -> PyResult<bool> {
698        let payload = py_to_nt_payload(nt)?;
699        let store = self.inner.clone();
700        Ok(block_on_py(py, store.put_nt(&name, payload)))
701    }
702
703    /// List all PV names in the store.
704    fn pv_names(&self, py: Python<'_>) -> PyResult<Vec<String>> {
705        let store = self.inner.clone();
706        Ok(block_on_py(py, store.pv_names()))
707    }
708
709    fn __repr__(&self, py: Python<'_>) -> String {
710        let store = self.inner.clone();
711        let n = block_on_py(py, store.pv_names()).len();
712        format!("<spvirit.Store ({n} PVs)>")
713    }
714}
715
716// ─── Helpers ─────────────────────────────────────────────────────────────────
717
718/// Convert a Python value to a [`PvValue`].
719///
720/// Scalars (bool, int, float, str) become `PvValue::Scalar`.
721/// Lists become `PvValue::ScalarArray`.
722fn py_to_pv_value(obj: &Bound<'_, PyAny>) -> PyResult<spvirit_types::PvValue> {
723    if let Ok(list) = obj.downcast::<pyo3::types::PyList>() {
724        let arr = py_to_scalar_array(list.as_any())?;
725        Ok(spvirit_types::PvValue::ScalarArray(arr))
726    } else {
727        let sv = py_to_scalar(obj)?;
728        Ok(spvirit_types::PvValue::Scalar(sv))
729    }
730}
spvirit/server.rs

spvirit/
server.rs
