capva 0.1.2

EPICS PV client supporting both Channel Access and PV Access

capva

Unified Python client for EPICS PVs over Channel Access (CA) and PV Access (PVA).

capva wraps pyepics and p4p behind one API. Reads and monitors return a structured PVData model; JSON/Web payloads are built with PVData.to_dict() (optional base64 array encoding).

Developed from the weiss project — capva builds on that codebase and refactors the PV client layer into a standalone Python library with a unified CA/PVA API and structured PVData.

flowchart TB
  subgraph capva["capva — standalone PV client library"]
    api["PV, PVPool, tools"]
    prov["CA_PV, PVA_PV"]
    data["pv_parser, PVData"]
  end

  subgraph drivers["protocol drivers"]
    direction LR
    pyepics["pyepics (CA)"]
    p4p["p4p (PVA)"]
  end

  epics["EPICS IOC"]

  prov --> pyepics
  prov --> p4p
  pyepics --> epics
  p4p --> epics

Features

• Single API for CA and PVA — One client for Channel Access and PV Access; capva picks the backend from the PV name so application code does not split into separate CA/PVA paths.
• Unified PVData model — Every read returns the same structured snapshot (value, alarm, timeStamp, display, control, …). Monitor callbacks return value, alarm, and timeStamp by default; pass include_metadata=True to also parse display, control, and valueAlarm from monitor events (no extra CA/PVA GET).
• Public interfaces: PV, PVPool, and procedural tools — Use the object API for multi-step work on one connection, PVPool when several callers share a PV, or one-shot pvget / pvput / pvinfo / pvmonitor when a script only needs a single operation.
• Reference-counted PVPool — getPV / releasePV reuse one connection per PV name; the channel closes when the last reference is released.
• Protocol prefixes — ca://… for Channel Access, pva://… for PV Access, or no prefix to default to CA.

Requirements

• Python ≥ 3.10

Installation

From PyPI:

pip install capva

From a checkout (development):

pip install -e .

Quick start

Examples use pva://calcExample; switch to ca://… or a bare name (CA default) as needed.

Procedural API

import time

from capva import PVData, pvget, pvmonitor

PV_NAME = "pva://calcExample"

data = pvget(PV_NAME)
print(data.value)

def on_update(data: PVData) -> None:
    if data.is_disconnected():
        print(f"{data.pvName} disconnected")
    else:
        print(data.value)

session = pvmonitor(PV_NAME, on_update)
try:
    time.sleep(30)
finally:
    session.close()

Object API

import time

from capva import PV, PVData

PV_NAME = "pva://calcExample"

pv = PV(PV_NAME)
handle = None
try:
    data = pv.get()
    print(data.value)

    def on_update(data: PVData) -> None:
        if data.is_disconnected():
            print(f"{data.pvName} disconnected")
        else:
            print(data.value)

    handle = pv.monitor(on_update)
    # Or include display/control/valueAlarm in each monitor callback:
    # handle = pv.monitor(on_update, include_metadata=True)
    time.sleep(30)
finally:
    if handle is not None:
        pv.clear_monitor(handle)
    pv.close()

PVPool

from capva import PVPool

PV_NAME = "pva://calcExample"

pv1 = PVPool.getPV(PV_NAME)
pv2 = PVPool.getPV(PV_NAME)
try:
    print(pv1 is pv2)  # same pooled instance
    print(pv1.get().value)
finally:
    PVPool.releasePV(pv1)
    PVPool.releasePV(pv2)

PVData.to_dict modes

mode	Use case
"full"	Complete snapshot (value, alarm, timeStamp, display, control, …)
"update"	Monitor/Web push (value, alarm, timeStamp; no metadata unless parsed via include_metadata)
"metadata"	display / control / valueAlarm only

Set base64_encode=True on "full" or "update" to emit b64arr / b64dtype instead of a numeric array value.

Project layout

src/capva/
  pv.py, pv_data.py      # Public PV + PVData model
  pv_parser.py           # CA/PVA → PVData
  tools.py               # pvget, pvput, pvinfo, pvmonitor
  pool.py                # PVPool
  providers/             # ca_pv, pva_pv
examples/
  tool_*.py              # Procedural tools (one-shot)
  pv_*.py                # PV class API
  pool_*.py              # PVPool (shared connections)
tests/                   # Unit tests (mocked; no IOC required)

Examples

Edit PV_NAME at the top of each script, then run against a real IOC:

# Procedural tools
python examples/tool_get.py
python examples/tool_info.py
python examples/tool_put.py
python examples/tool_monitor.py

# PV class
python examples/pv_get.py
python examples/pv_info.py
python examples/pv_put.py
python examples/pv_monitor.py

# PVPool
python examples/pool_get.py
python examples/pool_info.py
python examples/pool_put.py
python examples/pool_monitor.py

# Web JSON payload (wfExample waveform PV + Node.js)
python examples/encode_array.py
node examples/decode_array.js

License

MIT — see LICENSE.