rename Server/server to rvsttd

rvsttd/src/main.rs

@@ -0,0 +1,478 @@
+use anyhow::{anyhow, Result};
+use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
+use cpal::{SampleFormat, SampleRate};
+use std::collections::HashSet;
+use std::ffi::CStr;
+use std::io::{BufRead, BufReader, Write};
+use std::net::{TcpListener, TcpStream};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+include!(concat!(env!("OUT_DIR"), "/moonshine_bindings.rs"));
+
+const SAMPLE_RATE: i32 = 16000;
+const HEADER_VERSION: i32 = 30000;
+const ARCH: u32 = 5; // MOONSHINE_MODEL_ARCH_MEDIUM_STREAMING
+const BIND_ADDR: &str = "127.0.0.1:6996";
+const MAX_TEXT_BYTES: usize = 1380;
+
+// ─── helpers ──────────────────────────────────────────────────────────────
+
+fn ts() -> String {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default();
+    let secs = now.as_secs() % 86400;
+    let h = secs / 3600;
+    let m = (secs % 3600) / 60;
+    let s = secs % 60;
+    let ms = now.subsec_millis();
+    format!("{:02}:{:02}:{:02}.{:03}", h, m, s, ms)
+}
+
+fn log(msg: &str) {
+    eprintln!("[{}] {}", ts(), msg);
+}
+
+fn err_str(code: i32) -> String {
+    unsafe {
+        let s = moonshine_error_to_string(code);
+        if s.is_null() {
+            format!("error {}", code)
+        } else {
+            CStr::from_ptr(s).to_string_lossy().into_owned()
+        }
+    }
+}
+
+fn truncate_to_word(text: &str, max_bytes: usize) -> &str {
+    if text.len() <= max_bytes {
+        return text;
+    }
+    let cut = &text[..max_bytes.min(text.len())];
+    match cut.rfind(' ') {
+        Some(pos) => &text[..pos],
+        None => cut,
+    }
+}
+
+fn line_text(line: &transcript_line_t) -> String {
+    if line.text.is_null() {
+        return String::new();
+    }
+    unsafe { CStr::from_ptr(line.text) }
+        .to_string_lossy()
+        .into_owned()
+}
+
+// ─── shared state ─────────────────────────────────────────────────────────
+
+struct Shared {
+    writer: Mutex<TcpStream>,
+    session_id: u64,
+    transcriber_handle: i32,
+}
+
+impl Shared {
+    fn send_msg(&self, prefix: &str, text: &str) {
+        let text = truncate_to_word(text, MAX_TEXT_BYTES);
+        let line = if text.is_empty() {
+            format!("{} {}\n", prefix, self.session_id)
+        } else {
+            format!("{} {} {}\n", prefix, self.session_id, text)
+        };
+        let mut writer = self.writer.lock().unwrap();
+        match writer.write_all(line.as_bytes()) {
+            Ok(_) => log(&format!("TX {} {} {}", prefix, self.session_id, text)),
+            Err(e) => log(&format!("TX failed: {}", e)),
+        }
+    }
+}
+
+// ─── session ──────────────────────────────────────────────────────────────
+
+struct Session {
+    shared: Arc<Shared>,
+    stop_signal: Arc<AtomicBool>,
+    aborted: Arc<AtomicBool>,
+    transcriber: thread::JoinHandle<()>,
+    cpal_stream: cpal::Stream,
+    stream_handle: i32,
+}
+
+impl Session {
+    fn stop(self) {
+        self.stop_signal.store(true, Ordering::SeqCst);
+        drop(self.cpal_stream);
+        self.transcriber.join().ok();
+        unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) };
+    }
+
+    fn abort(self) {
+        self.stop_signal.store(true, Ordering::SeqCst);
+        self.aborted.store(true, Ordering::SeqCst);
+        drop(self.cpal_stream);
+        self.transcriber.join().ok();
+        unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) };
+    }
+}
+
+fn start_session(shared: Arc<Shared>) -> Option<Session> {
+    let stream_handle = unsafe { moonshine_create_stream(shared.transcriber_handle, 0) };
+    if stream_handle < 0 {
+        log(&format!("create_stream failed: {}", err_str(stream_handle)));
+        return None;
+    }
+    let rc = unsafe { moonshine_start_stream(shared.transcriber_handle, stream_handle) };
+    if rc != 0 {
+        log(&format!("start_stream failed: {}", err_str(rc)));
+        unsafe { moonshine_free_stream(shared.transcriber_handle, stream_handle) };
+        return None;
+    }
+
+    let audio_buf: Arc<Mutex<Vec<f32>>> = Arc::new(Mutex::new(Vec::new()));
+    let stop_signal = Arc::new(AtomicBool::new(false));
+    let aborted = Arc::new(AtomicBool::new(false));
+
+    let cpal_stream = match start_cpal(audio_buf.clone(), stop_signal.clone()) {
+        Ok(s) => s,
+        Err(e) => {
+            log(&format!("cpal failed: {}", e));
+            unsafe { moonshine_free_stream(shared.transcriber_handle, stream_handle) };
+            return None;
+        }
+    };
+
+    let shared_clone = shared.clone();
+    let stop_signal_clone = stop_signal.clone();
+    let aborted_clone = aborted.clone();
+
+    let transcriber = thread::spawn(move || {
+        transcriber_loop(shared_clone, audio_buf, stop_signal_clone, aborted_clone, stream_handle);
+    });
+
+    Some(Session {
+        shared,
+        stop_signal,
+        aborted,
+        transcriber,
+        cpal_stream,
+        stream_handle,
+    })
+}
+
+fn transcriber_loop(
+    shared: Arc<Shared>,
+    audio_buf: Arc<Mutex<Vec<f32>>>,
+    stop_signal: Arc<AtomicBool>,
+    aborted: Arc<AtomicBool>,
+    stream_handle: i32,
+) {
+    let handle = shared.transcriber_handle;
+    let mut sent_ids: HashSet<u64> = HashSet::new();
+
+    while !stop_signal.load(Ordering::SeqCst) {
+        let chunk = {
+            let mut buf = audio_buf.lock().unwrap();
+            if buf.is_empty() {
+                drop(buf);
+                thread::sleep(Duration::from_millis(5));
+                continue;
+            }
+            std::mem::take(&mut *buf)
+        };
+
+        unsafe {
+            moonshine_transcribe_add_audio_to_stream(
+                handle, stream_handle,
+                chunk.as_ptr(), chunk.len() as u64,
+                SAMPLE_RATE, 0,
+            );
+        }
+
+        let mut t_ptr: *mut transcript_t = std::ptr::null_mut();
+        let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) };
+        if rc != 0 || t_ptr.is_null() {
+            continue;
+        }
+
+        send_new_segments(&shared, t_ptr, &mut sent_ids, "P");
+    }
+
+    // If aborted (new session took over), skip final flush entirely
+    if aborted.load(Ordering::SeqCst) {
+        unsafe { moonshine_stop_stream(handle, stream_handle) };
+        log(&format!("Session {} aborted, skipping final flush", shared.session_id));
+        return;
+    }
+
+    // Drain remaining audio
+    let remaining = {
+        let mut buf = audio_buf.lock().unwrap();
+        std::mem::take(&mut *buf)
+    };
+    if !remaining.is_empty() {
+        unsafe {
+            moonshine_transcribe_add_audio_to_stream(
+                handle, stream_handle,
+                remaining.as_ptr(), remaining.len() as u64,
+                SAMPLE_RATE, 0,
+            );
+        }
+    }
+
+    // Final flush
+    unsafe { moonshine_stop_stream(handle, stream_handle) };
+    let mut t_ptr: *mut transcript_t = std::ptr::null_mut();
+    let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) };
+
+    if rc == 0 && !t_ptr.is_null() {
+        let t = unsafe { &*t_ptr };
+        let mut new_segments: Vec<String> = Vec::new();
+
+        for i in 0..t.line_count as usize {
+            let line = unsafe { &*t.lines.add(i) };
+            if line.text.is_null() || line.is_complete == 0 {
+                continue;
+            }
+            if !sent_ids.insert(line.id) {
+                continue;
+            }
+            let text = line_text(line);
+            if !text.is_empty() {
+                new_segments.push(text);
+            }
+        }
+
+        if new_segments.is_empty() {
+            shared.send_msg("F", "");
+        } else {
+            let last = new_segments.len() - 1;
+            for (i, text) in new_segments.iter().enumerate() {
+                let prefix = if i == last { "F" } else { "P" };
+                shared.send_msg(prefix, text);
+            }
+        }
+    } else {
+        shared.send_msg("F", "");
+    }
+}
+
+fn send_new_segments(
+    shared: &Shared,
+    t_ptr: *const transcript_t,
+    sent_ids: &mut HashSet<u64>,
+    prefix: &str,
+) {
+    let t = unsafe { &*t_ptr };
+
+    for i in 0..t.line_count as usize {
+        let line = unsafe { &*t.lines.add(i) };
+        if line.text.is_null() || line.is_complete == 0 {
+            continue;
+        }
+        if !sent_ids.insert(line.id) {
+            continue;
+        }
+        let text = line_text(line);
+        if text.is_empty() {
+            continue;
+        }
+        shared.send_msg(prefix, &text);
+    }
+}
+
+// ─── cpal ─────────────────────────────────────────────────────────────────
+
+fn start_cpal(
+    audio_buf: Arc<Mutex<Vec<f32>>>,
+    stop_signal: Arc<AtomicBool>,
+) -> Result<cpal::Stream> {
+    let host = cpal::default_host();
+    let dev = host
+        .default_input_device()
+        .ok_or_else(|| anyhow!("no input device"))?;
+
+    let supported = dev
+        .supported_input_configs()?
+        .filter(|c| c.channels() <= 2 && c.min_sample_rate().0 <= 16000)
+        .min_by_key(|c| match c.sample_format() {
+            SampleFormat::F32 => 0,
+            SampleFormat::I16 => 1,
+            SampleFormat::U8 => 2,
+            _ => 99,
+        })
+        .ok_or_else(|| anyhow!("no suitable input config"))?;
+
+    let fmt = supported.sample_format();
+    let mut config = supported.with_max_sample_rate().config();
+    if config.channels > 1 {
+        config.channels = 1;
+    }
+    config.sample_rate = SampleRate(16000);
+
+    let err_fn = |e: cpal::StreamError| log(&format!("cpal error: {}", e));
+
+    let stream = match fmt {
+        SampleFormat::F32 => dev.build_input_stream(
+            &config,
+            move |data: &[f32], _: &_| {
+                if !stop_signal.load(Ordering::Relaxed) {
+                    audio_buf.lock().unwrap().extend_from_slice(data);
+                }
+            },
+            err_fn,
+            None,
+        )?,
+        SampleFormat::I16 => dev.build_input_stream(
+            &config,
+            move |data: &[i16], _: &_| {
+                if !stop_signal.load(Ordering::Relaxed) {
+                    audio_buf.lock().unwrap().extend(data.iter().map(|&x| x as f32 / 32768.0));
+                }
+            },
+            err_fn,
+            None,
+        )?,
+        SampleFormat::U8 => dev.build_input_stream(
+            &config,
+            move |data: &[u8], _: &_| {
+                if !stop_signal.load(Ordering::Relaxed) {
+                    audio_buf.lock().unwrap().extend(data.iter().map(|&x| (x as f32 - 128.0) / 128.0));
+                }
+            },
+            err_fn,
+            None,
+        )?,
+        _ => return Err(anyhow!("unsupported sample format {:?}", fmt)),
+    };
+
+    stream.play()?;
+    Ok(stream)
+}
+
+// ─── main ─────────────────────────────────────────────────────────────────
+
+fn main() -> Result<()> {
+    let mut args = std::env::args().skip(1);
+    let mut model_dir = String::from("../bench/medium-streaming-en");
+
+    while let Some(a) = args.next() {
+        match a.as_str() {
+            "--model-dir" | "-m" => {
+                model_dir = args.next().unwrap_or(model_dir);
+            }
+            "--help" | "-h" => {
+                println!("Usage: rvsttd [--model-dir DIR]");
+                println!("Listens on TCP {}", BIND_ADDR);
+                println!("Model: medium-streaming (Moonshine)");
+                return Ok(());
+            }
+            _ => return Err(anyhow!("unknown arg: {}", a)),
+        }
+    }
+
+    let model_path = std::fs::canonicalize(&model_dir)
+        .unwrap_or_else(|_| std::path::PathBuf::from(&model_dir));
+
+    log(&format!("Loading model from {}...", model_path.display()));
+    let c_dir = std::ffi::CString::new(model_path.to_str().unwrap()).unwrap();
+    let transcriber_handle = unsafe {
+        moonshine_load_transcriber_from_files(
+            c_dir.as_ptr(),
+            ARCH,
+            std::ptr::null(),
+            0,
+            HEADER_VERSION,
+        )
+    };
+    if transcriber_handle < 0 {
+        return Err(anyhow!("failed to load model: {}", err_str(transcriber_handle)));
+    }
+    log(&format!("Model loaded (handle {})", transcriber_handle));
+
+    let listener = TcpListener::bind(BIND_ADDR)?;
+    log(&format!("STT server listening on TCP {}", BIND_ADDR));
+
+    let mut session_id_counter: u64 = 0;
+    let mut current_session: Option<Session> = None;
+
+    for stream in listener.incoming() {
+        let stream = match stream {
+            Ok(s) => s,
+            Err(e) => {
+                log(&format!("accept failed: {}", e));
+                continue;
+            }
+        };
+
+        stream.set_nodelay(true).ok();
+        log(&format!("Client connected: {}", stream.peer_addr().unwrap_or_default()));
+
+        let writer_stream = stream.try_clone()?;
+        let reader = BufReader::new(stream);
+
+        for line in reader.lines() {
+            let line = match line {
+                Ok(l) => l,
+                Err(_) => break,
+            };
+
+            let line = line.trim();
+            log(&format!("RX {}", line));
+
+            // ON <session>
+            if let Some(rest) = line.strip_prefix("ON ") {
+                let new_session_id: u64 = rest.parse().unwrap_or(0);
+
+                if let Some(s) = current_session.take() {
+                    log(&format!("Aborting session {} for new session {}", s.shared.session_id, new_session_id));
+                    s.abort();
+                }
+
+                session_id_counter = new_session_id;
+                let shared = Arc::new(Shared {
+                    writer: Mutex::new(writer_stream.try_clone()?),
+                    session_id: session_id_counter,
+                    transcriber_handle,
+                });
+
+                log(&format!("PTT on session {}", session_id_counter));
+                match start_session(shared) {
+                    Some(s) => current_session = Some(s),
+                    None => log("Failed to start session"),
+                }
+            }
+            // OFF <session>
+            else if let Some(rest) = line.strip_prefix("OFF ") {
+                let off_session: u64 = rest.parse().unwrap_or(0);
+
+                if let Some(s) = current_session.as_ref() {
+                    if s.shared.session_id == off_session {
+                        log(&format!("OFF session {}", off_session));
+                        if let Some(s) = current_session.take() {
+                            s.stop();
+                        }
+                    } else {
+                        log(&format!("OFF session {} (stale, current={}), ignoring", off_session, s.shared.session_id));
+                    }
+                } else {
+                    log(&format!("OFF session {} (no active session), ignoring", off_session));
+                }
+            }
+            else {
+                log(&format!("Unknown command: {}", line));
+            }
+        }
+
+        log("Client disconnected");
+
+        if let Some(s) = current_session.take() {
+            s.abort();
+        }
+    }
+
+    Ok(())
+}