WIP push-to-talk Letta chat frontend
display transcription in frontend
+199
-144
+13
-3
src-tauri/src/cartesia/commands.rs
+13
-3
src-tauri/src/cartesia/commands.rs
···
1
-
use crate::state::AppState;
1
+
use tauri::ipc::Channel;
2
+
3
+
use crate::{cartesia::stt::TranscriptionWord, state::AppState};
2
4
3
5
#[tauri::command]
4
-
pub async fn start_stt(state: tauri::State<'_, AppState>) -> Result<(), ()> {
5
-
state.stt_manager.transcribe().await;
6
+
pub async fn start_stt(
7
+
state: tauri::State<'_, AppState>,
8
+
on_event: Channel<TranscriptionWord>,
9
+
) -> Result<(), ()> {
10
+
let mut rec = state.stt_manager.transcribe().await;
11
+
12
+
while let Some(word) = rec.recv().await {
13
+
on_event.send(word).expect("failed to send word to client");
14
+
}
15
+
6
16
Ok(())
7
17
}
8
18
+147
-134
src-tauri/src/cartesia/stt.rs
+147
-134
src-tauri/src/cartesia/stt.rs
···
2
2
use crate::devices::input::InputDeviceManager;
3
3
use dasp::Signal;
4
4
use futures_util::{
5
-
future::join,
6
5
stream::{SplitSink, SplitStream},
7
6
SinkExt, StreamExt,
8
7
};
9
-
use serde::Deserialize;
8
+
use serde::{Deserialize, Serialize};
10
9
use std::sync::Arc;
11
-
use tauri::async_runtime::{Mutex, RwLock};
10
+
use tauri::async_runtime::{channel, spawn, Mutex, Receiver, RwLock, Sender};
12
11
use tokio::net::TcpStream;
13
12
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
14
13
use tungstenite::{Bytes, Error, Message, Utf8Bytes};
···
29
28
input: Arc<InputDeviceManager>,
30
29
}
31
30
32
-
#[derive(Deserialize, Debug)]
33
-
struct TranscriptionWord {
31
+
#[derive(Deserialize, Debug, Serialize)]
32
+
pub struct TranscriptionWord {
34
33
word: String,
35
34
start: f32,
36
35
end: f32,
37
36
}
38
37
39
38
#[derive(Deserialize, Debug)]
40
-
#[serde(untagged)]
41
-
enum TranscriptionMessage {
39
+
pub enum MessageType {
40
+
#[serde(rename = "transcript")]
41
+
Transcript,
42
+
#[serde(rename = "error")]
43
+
Error,
44
+
#[serde(rename = "done")]
45
+
Done,
46
+
#[serde(rename = "flush_done")]
47
+
FlushDone,
48
+
}
49
+
50
+
#[derive(Deserialize, Debug)]
51
+
#[serde(tag = "type", rename_all = "snake_case")]
52
+
pub enum TranscriptionMessage {
42
53
Transcript {
43
-
#[serde(rename = "type")]
44
-
message_type: String,
45
54
request_id: String,
46
55
is_final: bool,
47
56
text: String,
48
57
duration: Option<f32>,
49
58
language: Option<String>,
50
-
words: Vec<TranscriptionWord>,
59
+
words: Option<Vec<TranscriptionWord>>,
51
60
},
52
61
Error {
53
-
#[serde(rename = "type")]
54
-
message_type: String,
55
62
message: String,
56
63
request_id: Option<String>,
57
64
},
58
65
Done {
59
-
#[serde(rename = "type")]
60
-
message_type: String,
66
+
request_id: String,
67
+
},
68
+
FlushDone {
61
69
request_id: String,
62
70
},
63
71
}
···
71
79
}
72
80
}
73
81
74
-
async fn handle_messages(
75
-
&self,
76
-
mut reader: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
77
-
) {
78
-
println!("handling messages");
79
-
80
-
while let Some(message) = reader.next().await {
81
-
match message {
82
-
Ok(Message::Text(msg)) => {
83
-
println!("got message: {}", msg);
84
-
85
-
match serde_json::from_str::<TranscriptionMessage>(msg.as_str()) {
86
-
Ok(TranscriptionMessage::Transcript {
87
-
message_type,
88
-
request_id,
89
-
is_final,
90
-
text,
91
-
duration,
92
-
language,
93
-
words,
94
-
}) => {
95
-
println!("transcribed: {:?}", text);
96
-
}
97
-
Ok(TranscriptionMessage::Done {
98
-
message_type,
99
-
request_id,
100
-
}) => {
101
-
println!("recieved done message, stopping");
102
-
self.stop_transcription().await;
103
-
break;
104
-
}
105
-
Ok(msg) => {
106
-
println!("received a message: {:?}", msg)
107
-
}
108
-
Err(err) => {
109
-
eprintln!("failed to parse message: {}", err);
110
-
self.stop_transcription().await;
111
-
break;
112
-
}
113
-
}
114
-
}
115
-
Ok(msg) => {
116
-
println!("got non-text message: {}", msg);
117
-
}
118
-
Err(Error::ConnectionClosed) | Err(Error::AlreadyClosed) => {
119
-
eprintln!("connection closed");
120
-
break;
121
-
}
122
-
Err(err) => {
123
-
eprintln!("err with connection: {}", err);
124
-
break;
125
-
}
126
-
}
127
-
}
128
-
}
129
-
130
-
async fn send_frames(
131
-
&self,
132
-
signal: impl Signal<Frame = i16>,
133
-
writer: Arc<Mutex<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>,
134
-
) {
135
-
let mut w = writer.lock().await;
136
-
let mut buffer = Vec::with_capacity(3200);
137
-
138
-
for frame in signal.until_exhausted() {
139
-
let status = self.status.read().await;
140
-
141
-
match *status {
142
-
SttStatus::Connected => {
143
-
buffer.push(frame.to_le_bytes());
144
-
145
-
if buffer.len() == buffer.capacity() {
146
-
w.send(tungstenite::Message::Binary(Bytes::from_iter(
147
-
buffer.iter().flat_map(|f| *f),
148
-
)))
149
-
.await
150
-
.expect("failed to send binary frame message to STT");
151
-
152
-
buffer.clear();
153
-
}
154
-
}
155
-
SttStatus::Flushing => {
156
-
println!("flushing audio data");
157
-
158
-
w.send(tungstenite::Message::Binary(Bytes::from_iter(
159
-
buffer.iter().flat_map(|f| *f),
160
-
)))
161
-
.await
162
-
.expect("failed to send binary frame message to STT");
163
-
164
-
w.send(Message::Text(Utf8Bytes::from_static("done")))
165
-
.await
166
-
.expect("failed to send done message");
167
-
168
-
let _ = w.flush().await;
169
-
break;
170
-
}
171
-
_ => (),
172
-
}
173
-
}
174
-
}
175
-
176
82
/// Begins transcribing text via Cartesia. Blocks until `stop_transcription` is called
177
-
pub async fn transcribe(&self) {
178
-
println!("starting transcription");
179
-
83
+
pub async fn transcribe(&self) -> Receiver<TranscriptionWord> {
180
84
{
181
85
let mut status = self.status.write().await;
182
86
*status = SttStatus::Opening;
···
184
88
185
89
let input = self.input.start_listening().await;
186
90
let stream = self.client.open_stt_connection().await;
187
-
let (tx, rx) = stream.split();
188
-
let writer = Arc::new(Mutex::new(tx));
91
+
let (socket_in, socket_out) = stream.split();
92
+
let (fn_in, fn_out) = channel::<TranscriptionWord>(500);
93
+
let writer = Arc::new(Mutex::new(socket_in));
189
94
190
95
{
191
96
let mut status = self.status.write().await;
192
97
*status = SttStatus::Connected;
193
98
}
194
99
195
-
// Handle incoming messages
196
-
let read = self.handle_messages(rx);
197
-
let write = self.send_frames(input, writer.clone());
100
+
spawn(handle_messages(
101
+
self.status.clone(),
102
+
self.input.clone(),
103
+
socket_out,
104
+
writer.clone(),
105
+
fn_in,
106
+
));
107
+
spawn(send_frames(self.status.clone(), input, writer.clone()));
198
108
199
-
join(read, write).await;
200
-
201
-
writer
202
-
.lock()
203
-
.await
204
-
.send(Message::Close(None))
205
-
.await
206
-
.expect("failed to close socket");
109
+
return fn_out;
207
110
}
208
111
209
112
async fn flush_audio(&self) {
···
213
116
214
117
/// Terminates the microphone signal and halts the transcription processes
215
118
pub async fn stop_transcription(&self) {
216
-
println!("stopping transcription");
217
-
218
119
self.flush_audio().await;
219
120
self.input.stop_listening().await;
220
121
}
221
122
}
123
+
124
+
async fn handle_messages(
125
+
status: Arc<RwLock<SttStatus>>,
126
+
input: Arc<InputDeviceManager>,
127
+
mut socket_reader: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
128
+
socket_writer: Arc<Mutex<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>,
129
+
out_writer: Sender<TranscriptionWord>,
130
+
) {
131
+
while let Some(message) = socket_reader.next().await {
132
+
match message {
133
+
Ok(Message::Text(msg)) => {
134
+
match serde_json::from_str::<TranscriptionMessage>(msg.as_str()) {
135
+
Ok(TranscriptionMessage::Transcript {
136
+
request_id: _,
137
+
is_final: _,
138
+
text,
139
+
duration: _,
140
+
language: _,
141
+
words,
142
+
}) => match words {
143
+
Some(ws) => {
144
+
for w in ws {
145
+
out_writer.send(w).await.expect("failed to queue word")
146
+
}
147
+
}
148
+
None => (),
149
+
},
150
+
Ok(TranscriptionMessage::Done { request_id: _ }) => {
151
+
input.stop_listening().await;
152
+
socket_writer
153
+
.lock()
154
+
.await
155
+
.send(Message::Close(None))
156
+
.await
157
+
.expect("failed to close socket");
158
+
159
+
break;
160
+
}
161
+
Ok(msg) => {
162
+
println!("received a message: {:?}", msg)
163
+
}
164
+
Err(err) => {
165
+
eprintln!("failed to parse message: {}", err);
166
+
167
+
{
168
+
let mut status = status.write().await;
169
+
*status = SttStatus::Flushing;
170
+
}
171
+
172
+
input.stop_listening().await;
173
+
break;
174
+
}
175
+
}
176
+
}
177
+
Ok(msg) => {
178
+
println!("got non-text message: {}", msg);
179
+
}
180
+
Err(Error::ConnectionClosed) | Err(Error::AlreadyClosed) => {
181
+
eprintln!("connection closed");
182
+
break;
183
+
}
184
+
Err(err) => {
185
+
eprintln!("err with connection: {}", err);
186
+
break;
187
+
}
188
+
}
189
+
}
190
+
}
191
+
192
+
async fn send_frames(
193
+
status: Arc<RwLock<SttStatus>>,
194
+
signal: impl Signal<Frame = i16>,
195
+
writer: Arc<Mutex<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>,
196
+
) {
197
+
let mut w = writer.lock().await;
198
+
let mut buffer = Vec::with_capacity(3200);
199
+
200
+
for frame in signal.until_exhausted() {
201
+
let status = status.read().await;
202
+
203
+
match *status {
204
+
SttStatus::Connected => {
205
+
buffer.push(frame.to_le_bytes());
206
+
207
+
if buffer.len() == buffer.capacity() {
208
+
w.send(tungstenite::Message::Binary(Bytes::from_iter(
209
+
buffer.iter().flat_map(|f| *f),
210
+
)))
211
+
.await
212
+
.expect("failed to send binary frame message to STT");
213
+
214
+
buffer.clear();
215
+
}
216
+
}
217
+
SttStatus::Flushing => {
218
+
w.send(tungstenite::Message::Binary(Bytes::from_iter(
219
+
buffer.iter().flat_map(|f| *f),
220
+
)))
221
+
.await
222
+
.expect("failed to send binary frame message to STT");
223
+
224
+
w.send(Message::Text(Utf8Bytes::from_static("done")))
225
+
.await
226
+
.expect("failed to send done message");
227
+
228
+
let _ = w.flush().await;
229
+
break;
230
+
}
231
+
_ => (),
232
+
}
233
+
}
234
+
}
-2
src-tauri/src/devices/input.rs
-2
src-tauri/src/devices/input.rs
···
38
38
}
39
39
40
40
pub async fn start_listening(&self) -> impl Signal<Frame = i16> {
41
-
println!("opening device stream");
42
41
let config = self
43
42
.device
44
43
.supported_input_configs()
···
76
75
}
77
76
78
77
pub async fn stop_listening(&self) {
79
-
println!("dropping device stream");
80
78
let mut s = self.stream.lock().await;
81
79
*s = None;
82
80
}
+39
-5
src/routes/+page.svelte
+39
-5
src/routes/+page.svelte
···
1
1
<script lang="ts">
2
-
import { invoke } from "@tauri-apps/api/core";
2
+
import { invoke, Channel } from "@tauri-apps/api/core";
3
+
4
+
interface TranscriptionWord {
5
+
word: string;
6
+
start: number;
7
+
end: number;
8
+
}
3
9
10
+
let isRecording = $state(false);
11
+
let draft = $state("");
12
+
let history = $state(new Array<string>())
4
13
let keyInvocation = $state(
5
14
invoke("has_secret", { name: "cartesia_api_key" }),
6
15
);
7
-
let isRecording = $state(false);
8
16
9
-
$effect(() => {
10
-
if (isRecording) invoke("start_stt");
11
-
else invoke("stop_stt");
17
+
$effect(async () => {
18
+
if (isRecording) {
19
+
const onEvent = new Channel<TranscriptionWord>()
20
+
21
+
onEvent.onmessage = word => {
22
+
draft += word.word
23
+
}
24
+
25
+
invoke("start_stt", { onEvent });
26
+
} else {
27
+
await invoke("stop_stt");
28
+
29
+
if (draft) {
30
+
history.unshift(draft)
31
+
draft = ""
32
+
}
33
+
}
12
34
});
13
35
</script>
14
36
···
68
90
<p>{err}</p>
69
91
{/await}
70
92
</div>
93
+
94
+
{#if draft}
95
+
<div class="bg-rose-pine-surface bg-rose-pine-subtle italic rounded-md p-3 flex gap-3 shadow-xl">
96
+
{draft}
97
+
</div>
98
+
{/if}
99
+
100
+
{#each history as text}
101
+
<div class="bg-rose-pine-base rounded-md p-3 flex gap-3 shadow-xl">
102
+
{text}
103
+
</div>
104
+
{/each}