spvirit-py 0.1.7

Python bindings for spvirit PVAccess client and server.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

//! Python exception hierarchy mirroring `PvGetError`.
//!
//! ```text
//! SpviritError(Exception)
//! ├── TimeoutError
//! ├── SearchError
//! ├── ProtocolError
//! ├── DecodeError
//! └── IoError
//! ```

use pyo3::prelude::*;
use pyo3::create_exception;
use spvirit_client::PvGetError;

create_exception!(spvirit, SpviritError, pyo3::exceptions::PyException);
create_exception!(spvirit, TimeoutError, SpviritError);
create_exception!(spvirit, SearchError, SpviritError);
create_exception!(spvirit, ProtocolError, SpviritError);
create_exception!(spvirit, DecodeError, SpviritError);
create_exception!(spvirit, IoError, SpviritError);

pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add("SpviritError", m.py().get_type::<SpviritError>())?;
    m.add("TimeoutError", m.py().get_type::<TimeoutError>())?;
    m.add("SearchError", m.py().get_type::<SearchError>())?;
    m.add("ProtocolError", m.py().get_type::<ProtocolError>())?;
    m.add("DecodeError", m.py().get_type::<DecodeError>())?;
    m.add("IoError", m.py().get_type::<IoError>())?;
    Ok(())
}

/// Convert a `PvGetError` into the matching Python exception.
pub fn to_py_err(e: PvGetError) -> PyErr {
    match e {
        PvGetError::Io(e) => IoError::new_err(e.to_string()),
        PvGetError::Timeout(ctx) => TimeoutError::new_err(ctx.to_string()),
        PvGetError::Search(ctx) => SearchError::new_err(ctx.to_string()),
        PvGetError::Protocol(ctx) => ProtocolError::new_err(ctx),
        PvGetError::Decode(ctx) => DecodeError::new_err(ctx),
    }
}

/// Convert a raw `std::io::Error` into the matching Python exception.
#[allow(dead_code)]
pub fn io_to_py_err(e: std::io::Error) -> PyErr {
    IoError::new_err(e.to_string())
}

/// Convert a codec / decode string error to `DecodeError`.
pub fn decode_msg_to_py_err(msg: impl Into<String>) -> PyErr {
    DecodeError::new_err(msg.into())
}

/// Convert a protocol string error to `ProtocolError`.
pub fn protocol_msg_to_py_err(msg: impl Into<String>) -> PyErr {
    ProtocolError::new_err(msg.into())
}