Custom data sources

✅ Verified · multi_source.rs · wildcard_source.rs · json_source.rs · rpc_source.rs · demo_source_multi.py · demo_source_wildcard.py · demo_source_sensor.py · demo_source_rpc.py · check docs_verify ·

The badge reports the whole docs-verify suite, not this chapter alone.

What you'll build

A PV provider that is not a record store at all — PVs backed by a file, by another process, by a naming convention, or computed on demand.

Everything so far has used the built-in record store: you declare PVs up front and the server holds their values. A source replaces that with your own code. The server asks "do you own this name?" and, if you say yes, routes every GET, PUT and subscribe to you.

The Source trait

#![allow(unused)]
fn main() {
fn claim(&self, name: &str)   -> Option<PvInfo>;   // do I own this name?
fn get(&self, name: &str)     -> Option<NtPayload>;
fn put(&self, name, value)    -> Result<Vec<(String, NtPayload)>, String>;
fn subscribe(&self, name)     -> Option<Receiver<NtPayload>>;
fn rpc(&self, name, args)     -> Result<NtPayload, String>;   // has a default
fn names(&self)               -> Vec<String>;
}

(Shown with the Pin<Box<dyn Future<...>>> wrappers elided — the trait is object-safe rather than async fn, so every method returns a boxed future. See spvirit-server/src/pvstore.rs:55.)

claim is the interesting one. It runs on every channel search, and returning Some(PvInfo) commits you to serving that name.

In Python there is no trait to implement — a source is any object with the matching methods, checked by duck typing:

class MySource:
    def claim(self, name): ...        # -> PvInfo | None
    def get(self, name): ...          # -> NtScalar/NtScalarArray/... | None
    def put(self, name, value): ...   # -> payload, or raise to reject
    def rpc(self, name, args): ...    # optional
    def names(self): ...              # -> list[str]
    def on_start(self, notifier): ... # optional: stash the notifier

Two differences from Rust worth knowing up front. on_start has no Rust counterpart — it is how a Python source gets the Notifier it needs to push monitor updates. And subscribe is not part of the Python protocol: define it and it is ignored (spvirit-py/src/source.rs:535). Monitors are driven by notifier.notify(name, payload) instead.

Priority and the registry

Sources are registered with an integer order. Lower is checked first, and the built-in record store sits at 0:

#![allow(unused)]
fn main() {
    let server = PvaServer::builder()
        .ai("SIM:COUNTER", 0.0)
        // ConstSource at order -10 — checked before the built-in store
        .source("constants", -10, Arc::new(ConstSource::new()))
        // ComputedSource at order 10 — checked after the built-in store
        .source("computed", 10, Arc::new(ComputedSource))
        .build();
}

    # Lower order is asked first; the built-in record store sits at 0.
    # Fallback claims every name, so it goes last.
    server = (
        spvirit.ServerBuilder()
        .ai("BLT:X", 3.14)                        # built-in store, order 0
        .add_source("fast", 10, FastCache())
        .add_source("fallback", 100, Fallback())
        .build()
    )

So -10 shadows the built-in store, and 10 is a fallback for names it does not know. This is the whole resolution model: first claim wins.

Claiming by naming convention

A source can serve PVs that were never declared. This one owns everything starting with XYZ: and creates each PV on first touch:

#![allow(unused)]
fn main() {
    fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
        let name = name.to_string();
        Box::pin(async move {
            if !self.matches(&name) {
                return None;
            }
            // Every wildcard PV is writable and uses F64
            Some(PvInfo {
                descriptor: f64_scalar_desc(),
                writable: true,
            })
        })
    }
}

The Python equivalent, claiming SCRATCH: instead — same three methods, plus the notifier that publishes each PUT to subscribers:

class WildcardSource:
    """Accept any PV under PREFIX and keep a per-name float value."""

    def __init__(self) -> None:
        self._values: dict[str, float] = {}
        self._notifier: spvirit.Notifier | None = None
        self._lock = threading.Lock()

    # Called by the server right after build() — stash the notifier.
    def on_start(self, notifier: spvirit.Notifier) -> None:
        self._notifier = notifier

    def claim(self, name: str):
        if not name.startswith(PREFIX):
            return None
        return spvirit.PvInfo.nt_scalar("double", writable=True)

    def get(self, name: str):
        if not name.startswith(PREFIX):
            return None
        with self._lock:
            val = self._values.setdefault(name, 0.0)
        return spvirit.NtScalar(val)

    def put(self, name: str, value):
        """value is a Python dict/value built from the PUT payload."""
        if not name.startswith(PREFIX):
            return None
        new_val = _coerce_float(value)
        with self._lock:
            self._values[name] = new_val
        # Publish the update to PVA monitor subscribers.
        if self._notifier is not None:
            self._notifier.notify(name, spvirit.NtScalar(new_val))
        # Return propagation list (only this PV changed).
        return spvirit.NtScalar(new_val)

    def names(self):
        # Report only the names we've seen so far. A dynamic namespace
        # cannot enumerate what does not exist yet; the PVs still serve.
        with self._lock:
            return list(self._values.keys())

$ spget XYZ:NEW
XYZ:NEW   0          # sprang into existence on the search

$ spput XYZ:NEW 42 && spget XYZ:NEW
XYZ:NEW  42

$ spget ABC:NEW
Error: Timeout("search response")    # nothing claims ABC:

Note the failure mode for an unclaimed name: a search timeout, not a "not found" error. Nothing answers the UDP search, so the client waits.

Backing PVs with a file

#![allow(unused)]
fn main() {
impl Source for JsonSource {
    fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
        let name = name.to_string();
        Box::pin(async move {
            if self.pvs.read().await.contains_key(&name) {
                Some(PvInfo {
                    descriptor: f64_desc(),
                    writable: true,
                })
            } else {
                None
            }
        })
    }

    fn get(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<NtPayload>> + Send + '_>> {
        let name = name.to_string();
        Box::pin(async move {
            let pvs = self.pvs.read().await;
            let val = *pvs.get(&name)?;
            Some(NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(
                val,
            ))))
        })
    }

    fn put(
        &self,
        name: &str,
        value: &DecodedValue,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, NtPayload)>, String>> + Send + '_>> {
        let name = name.to_string();
        let value = value.clone();
        Box::pin(async move {
            let new_val = match &value {
                DecodedValue::Float64(v) => *v,
                DecodedValue::Int32(v) => *v as f64,
                DecodedValue::Structure(fields) => fields
                    .iter()
                    .find(|(k, _)| k == "value")
                    .and_then(|(_, v)| match v {
                        DecodedValue::Float64(f) => Some(*f),
                        DecodedValue::Int32(i) => Some(*i as f64),
                        _ => None,
                    })
                    .ok_or("missing numeric 'value' field")?,
                _ => return Err("unsupported value type".to_string()),
            };

            {
                let mut pvs = self.pvs.write().await;
                if !pvs.contains_key(&name) {
                    return Err(format!("PV '{}' not found in JSON store", name));
                }
                pvs.insert(name.clone(), new_val);
            }

            // Persist to disk after every write
            self.persist().await;
            println!("[json_source] persisted {} = {}", name, new_val);

            let payload = NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(new_val)));
            Ok(vec![(name, payload)])
        })
    }
}

Values survive a restart because put writes the JSON file synchronously and the constructor reads it back:

$ spput JSON:SETPOINT_A 123.4
JSON:SETPOINT_A OK
# ... stop the server, start it again ...
$ spget JSON:SETPOINT_A
JSON:SETPOINT_A 123.4

Pushing monitor updates

A source with a value that changes on its own — a polled instrument, a subscription to some other system — needs to push. In Rust that is subscribe, returning a channel the server drains. In Python subscribe is ignored; you keep the Notifier from on_start and call it from whatever thread produces the data:

    # A Python source's `subscribe` is ignored: monitor events come from the
    # notifier the server hands to `on_start`. Nothing polls `get` on your
    # behalf, so a source that never notifies looks frozen to a subscriber.
    def _loop(self):
        t = 0.0
        while not self._stop.is_set():
            new = {
                "SENSOR:TEMP":     22.0 + 2.0 * math.sin(t),
                "SENSOR:PRESSURE": 1.0 + 0.1 * math.sin(t * 1.7),
                "SENSOR:FLOW":     5.0 + 1.0 * math.sin(t * 0.4),
            }
            with self._lock:
                self._values.update(new)
            # Publish updates — this is what delivers monitor events to clients.
            if self._notifier is not None:
                for name, v in new.items():
                    self._notifier.notify(name, spvirit.NtScalar(v))
            t += 0.2
            time.sleep(0.5)

RPC

rpc is the one trait method with a default implementation — it returns Err("RPC not supported"), so a source only opts in by overriding it:

#![allow(unused)]
fn main() {
    fn rpc(
        &self,
        name: &str,
        args: &DecodedValue,
    ) -> Pin<Box<dyn Future<Output = Result<NtPayload, String>> + Send + '_>> {
        let name = name.to_string();
        let args = args.clone();
        Box::pin(async move {
            if name != "RPC:add" {
                return Err(format!("unknown RPC channel '{}'", name));
            }

            let a = extract_f64(&args, "a").unwrap_or(0.0);
            let b = extract_f64(&args, "b").unwrap_or(0.0);
            let sum = a + b;
            println!("[rpc] {a} + {b} = {sum}");

            Ok(NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(
                sum,
            ))))
        })
    }
}

In Python it is likewise optional — a source without an rpc method simply has no RPC channel:

    # `rpc` is optional: a source that does not define it simply has no RPC.
    def rpc(self, name: str, args):
        if name != self.CHANNEL:
            raise RuntimeError(f"unknown RPC channel: {name}")
        # `args` is a Python dict built from the decoded request structure.
        a = _as_float(args.get("a", 0.0))
        b = _as_float(args.get("b", 0.0))
        return spvirit.NtScalar(a + b)

spvirit ships no general-purpose RPC client. Neither spvirit-client nor any of the CLI tools can call an arbitrary RPC channel — the only RPC in the client is an internal path used by pvlist. To exercise an RPC source, use p4p or pvxs:

from p4p.client.thread import Context
ctx = Context('pva')
print(ctx.rpc('RPC:add', {'a': 3.0, 'b': 4.0}))   # 7.0

Other shapes in the repo

The Python family is demo_source_*.py — sensor, async, multi, passthrough, aggregate, rpc, wildcard. demo_source_async.py is the one with no Rust counterpart here: it shows a source whose get is an async def, which the adapter awaits on the server's runtime.

What to notice

claim is on the hot path. It is called for every channel search from every client on the network. Keep it cheap — no I/O, no locks held across awaits. Do the expensive work in get.

names() drives splist. A source that returns an empty names() still serves its PVs; they just do not show up in listings. The wildcard source cannot enumerate what does not exist yet, which is the honest answer for a dynamic namespace.

Sources bypass the record layer entirely. No MDEL, no alarm computation, no scan, no .FIELD access — those are properties of RecordInstance, and a source does not have one. If you want deadbands, implement them in subscribe.

Returning Some from claim is a commitment. There is no way to un-claim afterwards; a subsequent get returning None surfaces to the client as an error rather than falling through to the next source.

Run it

cargo run -p spvirit-server --example multi_source
cargo run -p spvirit-server --example wildcard_source
cargo run -p spvirit-server --example json_source
cargo run -p spvirit-server --example rpc_source

python spvirit-py/examples/demo_source_multi.py
python spvirit-py/examples/demo_source_wildcard.py
python spvirit-py/examples/demo_source_sensor.py
python spvirit-py/examples/demo_source_rpc.py

Each is a server; drive it from a second terminal. multi_source registers several sources on one server, and splist shows them merged into one flat namespace — nothing in the listing says which source owns which PV:

$ splist 127.0.0.1:5075
COMPUTED:TIME
CONST:E
CONST:PI
SIM:COUNTER
__pvlist

$ spget CONST:PI
CONST:PI 2026-08-06 09:19:57.058 3.141593

$ spget COMPUTED:TIME
COMPUTED:TIME 2026-08-06 09:19:57.139 1786007997.139291

$ spget SIM:COUNTER
SIM:COUNTER 2026-08-06 09:19:56.859   3

wildcard_source claims a whole prefix, so the PV does not exist until you write to it:

$ spput XYZ:MyValue 42.0
XYZ:MyValue OK

$ spget XYZ:MyValue
XYZ:MyValue  42

$ spput XYZ:sensor/temp 21.5
XYZ:sensor/temp OK

$ splist 127.0.0.1:5075
STATIC:HEARTBEAT
XYZ:MyValue
XYZ:sensor/temp
__pvlist

Two things to notice. spget XYZ:MyValue prints no timestamp — the source returns a bare value and nothing stamped it, unlike a record. And splist only reports the names created so far; a wildcard source cannot enumerate an infinite namespace.

json_source writes through to disk, so the value survives a restart:

$ spput JSON:SETPOINT_A 123.4
JSON:SETPOINT_A OK

$ spget JSON:SETPOINT_A
JSON:SETPOINT_A 123.4

# stop the server with Ctrl-C, start it again

$ spget JSON:SETPOINT_A
JSON:SETPOINT_A 123.4

The server prints its side of that on startup:

[json_source] loaded 4 PVs from pvstore.json
JSON file-backed source server running on port 5075
  Persistent PVs: JSON:SETPOINT_A, JSON:SETPOINT_B, JSON:LIMIT_HI, JSON:LIMIT_LO
  In-memory PV:   SIM:HEARTBEAT
  Storage file:   pvstore.json

It creates pvstore.json in your working directory — delete it if you want to start from the defaults again.

rpc_source has no expected output here, because as noted above spvirit ships no RPC client; use p4p or pvcall from pvxs against it.

Next

A complete IOC.