lasprs 0.14.2

Library for Acoustic Signal Processing (Rust edition, with optional Python bindings via pyo3)
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

use crate::daq::{
    InQueues, InStreamData, InStreamMsg, InputStreamCommand, PreCaptureBuffer, StreamMetaData,
};
use crate::{daq::error::StreamMgrError, *};
use crossbeam::channel::{Receiver, Sender, unbounded};
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;

type Result<T> = std::result::Result<T, StreamMgrError>;

pub fn startInputStreamThread(
    meta: Arc<StreamMetaData>,
    rx: Receiver<InStreamMsg>,
    mut iqueues: InQueues,
) -> (
    JoinHandle<InQueues>,
    Sender<InputStreamCommand>,
    Receiver<Result<()>>,
) {
    // Bi-directional communication between input stream thread and stream manager
    let (commtx_ret, commrx) = unbounded();
    let (commtx, commrx_ret) = unbounded();

    // Compute pre-capture capacity once: 3 seconds worth of blocks.
    let precap_capacity = (PRECAP_DURATION.as_secs_f64() * *meta.samplerate
        / meta.framesPerBlock as f64)
        .ceil() as usize;

    let threadhandle = spawn(
        move || {
            let mut precap_buf: VecDeque<Arc<InStreamData>> =
                VecDeque::with_capacity(precap_capacity);
            // Tracks whether the buffer has been filled and wrapped at least once.
            // Only then we have a full 3 seconds of data.
            let mut precap_wrapped = false;

            'infy: loop {
                if let Ok(comm_msg) = commrx.try_recv() {
                    match comm_msg {
                        InputStreamCommand::AddInQueue(queue) => {
                            // 1st message: metadata
                            let precap = if precap_wrapped && !precap_buf.is_empty() {
                                PreCaptureBuffer::Loaded(precap_buf.iter().cloned().collect())
                            } else {
                                PreCaptureBuffer::NotLoaded
                            };

                            match queue.send(InStreamMsg::StreamStarted(meta.clone(), precap)) {
                                Ok(()) => {
                                    iqueues.push(queue);
                                }
                                Err(e) => {
                                    eprintln!("Cannot push to queue: {e}. Object destructed?");
                                }
                            }

                            commtx.send(Ok(())).unwrap();
                        }
                        InputStreamCommand::StopThread => {
                            sendMsgToAllQueuesRemoveUnused(
                                &mut iqueues,
                                InStreamMsg::StreamStopped,
                            );
                            commtx.send(Ok(())).unwrap();
                            break 'infy;
                        }
                    }
                }
                if let Ok(msg) = rx.recv_timeout(Duration::from_millis(10)) {
                    // Maintain the pre-capture buffer
                    if let InStreamMsg::InStreamData(ref data) = msg {
                        if precap_buf.len() >= precap_capacity {
                            precap_buf.pop_front();
                            precap_wrapped = true;
                        }
                        precap_buf.push_back(data.clone());
                    }
                    // Forward to all registered queues (unchanged)
                    sendMsgToAllQueuesRemoveUnused(&mut iqueues, msg);
                }
            }
            iqueues
        },
        ThreadPriority::High,
    );
    (threadhandle, commtx_ret, commrx_ret)
}

/// Send to all queues, remove queues that are disconnected when found out
/// on the way.
pub fn sendMsgToAllQueuesRemoveUnused(iqueues: &mut InQueues, msg: InStreamMsg) {
    // Loop over queues. Remove queues that error when we try to send
    // to them
    iqueues.retain(|q| q.try_send(msg.clone()).is_ok());
}