mod audio;
|
|
use std::{env, sync::Arc, time::Duration};
|
|
use anyhow::{Context, Result, anyhow};
|
use audio::load_pre_recorded_frames;
|
use libwebrtc::{
|
audio_source::native::NativeAudioSource,
|
prelude::{AudioFrame, AudioSourceOptions, RtcAudioSource},
|
};
|
use livekit::{
|
options::TrackPublishOptions,
|
prelude::{DataPacket, LocalAudioTrack, LocalTrack, ParticipantIdentity, Room, RoomOptions},
|
};
|
use reqwest::Client;
|
use serde::Serialize;
|
use tokio::time::sleep;
|
use tracing::{info, warn};
|
|
const TARGET_SAMPLE_RATE_HZ: u32 = 48_000;
|
const TARGET_NUM_CHANNELS: u16 = 1;
|
const TRACK_NAME: &str = "bot-main-audio";
|
const DEVICE_OUTPUT_TOPIC: &str = "device_output";
|
|
#[tokio::main(flavor = "multi_thread")]
|
async fn main() -> Result<()> {
|
init_tracing();
|
let config = Config::from_env()?;
|
let http = Client::builder()
|
.use_rustls_tls()
|
.build()
|
.context("failed to build helper http client")?;
|
|
let greeting_frames = load_pre_recorded_frames(
|
&http,
|
config.greeting_audio_file.as_deref(),
|
config.greeting_audio_url.as_deref(),
|
TARGET_SAMPLE_RATE_HZ,
|
TARGET_NUM_CHANNELS,
|
)
|
.await?;
|
|
let (room, _events) = Room::connect(
|
config.livekit_url.as_str(),
|
config.bot_token.as_str(),
|
RoomOptions::default(),
|
)
|
.await
|
.map_err(|error| anyhow!("failed to connect runtime helper to livekit: {error}"))?;
|
let room = Arc::new(room);
|
|
info!(
|
call_id = %config.call_id,
|
trace_id = %config.trace_id,
|
room_id = %config.room_id,
|
participant_alias = %redact(&config.bot_participant_identity),
|
greeting_source = %config.greeting_source,
|
"combrabo voice runtime helper connected"
|
);
|
|
let sink = BotAudioOutputSink::publish(
|
room.clone(),
|
&config.room_id,
|
&config.bot_participant_identity,
|
TRACK_NAME,
|
TARGET_SAMPLE_RATE_HZ,
|
u32::from(TARGET_NUM_CHANNELS),
|
)
|
.await?;
|
|
if config.device_output_smoke_enabled {
|
publish_device_output_smoke(room.as_ref(), &config).await?;
|
}
|
|
if let Some(frames) = greeting_frames {
|
info!(
|
call_id = %config.call_id,
|
frame_count = frames.len(),
|
greeting_source = %config.greeting_source,
|
"runtime helper starting greeting playback"
|
);
|
for frame in &frames {
|
sink.write_pcm_frame(frame).await?;
|
sleep(Duration::from_millis(20)).await;
|
}
|
sink.clear_buffer();
|
info!(
|
call_id = %config.call_id,
|
greeting_source = %config.greeting_source,
|
"runtime helper finished greeting playback"
|
);
|
} else {
|
warn!(
|
call_id = %config.call_id,
|
greeting_source = %config.greeting_source,
|
"runtime helper started without greeting audio; keeping published track alive"
|
);
|
}
|
|
wait_for_shutdown_signal().await?;
|
|
if let Err(error) = sink.close().await {
|
warn!(
|
call_id = %config.call_id,
|
error = %error,
|
"runtime helper failed to unpublish bot track cleanly"
|
);
|
}
|
if let Err(error) = room.close().await {
|
warn!(
|
call_id = %config.call_id,
|
error = %error,
|
"runtime helper failed to disconnect livekit room cleanly"
|
);
|
}
|
info!(call_id = %config.call_id, "runtime helper exited");
|
Ok(())
|
}
|
|
fn init_tracing() {
|
let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
|
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
|
let _ = tracing_subscriber::fmt()
|
.with_env_filter(env_filter)
|
.with_target(false)
|
.try_init();
|
}
|
|
struct Config {
|
call_id: String,
|
trace_id: String,
|
livekit_url: String,
|
room_id: String,
|
bot_token: String,
|
bot_participant_identity: String,
|
greeting_source: String,
|
greeting_audio_file: Option<String>,
|
greeting_audio_url: Option<String>,
|
role_id: String,
|
device_output_smoke_enabled: bool,
|
device_output_destination_identities: Vec<ParticipantIdentity>,
|
}
|
|
impl Config {
|
fn from_env() -> Result<Self> {
|
Ok(Self {
|
call_id: required_env("CV_CALL_ID")?,
|
trace_id: required_env("CV_TRACE_ID")?,
|
livekit_url: required_env("CV_LIVEKIT_URL")?,
|
room_id: required_env("CV_LIVEKIT_ROOM_ID")?,
|
bot_token: required_env("CV_LIVEKIT_BOT_TOKEN")?,
|
bot_participant_identity: required_env("CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY")?,
|
greeting_source: env::var("CV_GREETING_SOURCE")
|
.unwrap_or_else(|_| "no_audio".to_string()),
|
greeting_audio_file: optional_env("CV_GREETING_AUDIO_FILE"),
|
greeting_audio_url: optional_env("CV_GREETING_AUDIO_URL"),
|
role_id: env::var("CV_ROLE_ID").unwrap_or_else(|_| "90".to_string()),
|
device_output_smoke_enabled: bool_env("CV_DEVICE_OUTPUT_SMOKE_ENABLED"),
|
device_output_destination_identities: device_output_destinations(
|
optional_env("CV_DEVICE_OUTPUT_DESTINATION_IDENTITIES"),
|
optional_env("CV_LIVEKIT_USER_PARTICIPANT_IDENTITY"),
|
),
|
})
|
}
|
}
|
|
fn required_env(key: &str) -> Result<String> {
|
let value = env::var(key).with_context(|| format!("missing required env {key}"))?;
|
if value.trim().is_empty() {
|
return Err(anyhow!("required env {key} is blank"));
|
}
|
Ok(value)
|
}
|
|
fn optional_env(key: &str) -> Option<String> {
|
env::var(key).ok().and_then(|value| {
|
let trimmed = value.trim();
|
if trimmed.is_empty() {
|
None
|
} else {
|
Some(trimmed.to_string())
|
}
|
})
|
}
|
|
fn bool_env(key: &str) -> bool {
|
matches!(
|
env::var(key)
|
.unwrap_or_default()
|
.trim()
|
.to_ascii_lowercase()
|
.as_str(),
|
"1" | "true" | "yes" | "y" | "on"
|
)
|
}
|
|
fn device_output_destinations(
|
configured: Option<String>,
|
user_identity: Option<String>,
|
) -> Vec<ParticipantIdentity> {
|
let identities = configured
|
.filter(|value| !value.trim().is_empty())
|
.map(|value| {
|
value
|
.split(',')
|
.map(str::trim)
|
.filter(|identity| !identity.is_empty())
|
.map(ToOwned::to_owned)
|
.collect::<Vec<_>>()
|
})
|
.or_else(|| user_identity.map(|identity| vec![identity]))
|
.unwrap_or_default();
|
identities.into_iter().map(Into::into).collect()
|
}
|
|
fn redact(value: &str) -> String {
|
if value.len() <= 8 {
|
return "redacted".to_string();
|
}
|
format!("{}***{}", &value[..4], &value[value.len() - 4..])
|
}
|
|
#[derive(Serialize)]
|
struct DeviceOutputMessage {
|
#[serde(rename = "type")]
|
message_type: &'static str,
|
#[serde(rename = "schemaVersion")]
|
schema_version: &'static str,
|
#[serde(rename = "callId")]
|
call_id: String,
|
#[serde(rename = "roleId")]
|
role_id: String,
|
#[serde(rename = "traceId")]
|
trace_id: String,
|
#[serde(rename = "ackMode")]
|
ack_mode: &'static str,
|
#[serde(rename = "sensorInstructions")]
|
sensor_instructions: Vec<SensorInstruction>,
|
}
|
|
#[derive(Serialize)]
|
struct SensorInstruction {
|
#[serde(rename = "commandId")]
|
command_id: String,
|
#[serde(rename = "sensorType")]
|
sensor_type: &'static str,
|
#[serde(rename = "operationType")]
|
operation_type: &'static str,
|
step: i32,
|
#[serde(rename = "durationSec", skip_serializing_if = "Option::is_none")]
|
duration_sec: Option<i32>,
|
extension: String,
|
}
|
|
async fn publish_device_output_smoke(room: &Room, config: &Config) -> Result<()> {
|
let payload = build_device_output_smoke_payload(config);
|
let payload_json =
|
serde_json::to_vec(&payload).context("failed to serialize device output smoke payload")?;
|
let destination_count = config.device_output_destination_identities.len();
|
room.local_participant()
|
.publish_data(DataPacket {
|
reliable: true,
|
payload: payload_json,
|
topic: Some(DEVICE_OUTPUT_TOPIC.to_string()),
|
destination_identities: config.device_output_destination_identities.clone(),
|
})
|
.await
|
.map_err(|error| anyhow!("failed to publish livekit device output data: {error}"))?;
|
|
info!(
|
call_id = %config.call_id,
|
trace_id = %config.trace_id,
|
topic = DEVICE_OUTPUT_TOPIC,
|
reliable = true,
|
command_count = payload.sensor_instructions.len(),
|
destination_count,
|
"runtime helper published device output smoke data"
|
);
|
Ok(())
|
}
|
|
fn build_device_output_smoke_payload(config: &Config) -> DeviceOutputMessage {
|
DeviceOutputMessage {
|
message_type: DEVICE_OUTPUT_TOPIC,
|
schema_version: "1.0",
|
call_id: config.call_id.clone(),
|
role_id: config.role_id.clone(),
|
trace_id: config.trace_id.clone(),
|
ack_mode: "http",
|
sensor_instructions: vec![
|
SensorInstruction {
|
command_id: command_id(&config.call_id, 1),
|
sensor_type: "Vibrator",
|
operation_type: "VibratorStart",
|
step: 1,
|
duration_sec: None,
|
extension: r#"{"levelList":[{"level":1,"percent":"0.6"}]}"#.to_string(),
|
},
|
SensorInstruction {
|
command_id: command_id(&config.call_id, 2),
|
sensor_type: "Vibrator",
|
operation_type: "VibratorUp",
|
step: 1,
|
duration_sec: None,
|
extension: r#"{"delta":1}"#.to_string(),
|
},
|
SensorInstruction {
|
command_id: command_id(&config.call_id, 3),
|
sensor_type: "Pump",
|
operation_type: "JiaStart",
|
step: 1,
|
duration_sec: None,
|
extension: r#"{"levelList":[{"level":1,"percent":"0.5"}]}"#.to_string(),
|
},
|
SensorInstruction {
|
command_id: command_id(&config.call_id, 4),
|
sensor_type: "Heating",
|
operation_type: "HeatingStart",
|
step: 1,
|
duration_sec: Some(3),
|
extension: r#"{"target":"warm"}"#.to_string(),
|
},
|
],
|
}
|
}
|
|
fn command_id(call_id: &str, sequence: u8) -> String {
|
format!("{call_id}-device-smoke-{sequence:03}")
|
}
|
|
struct BotAudioOutputSink {
|
room: Arc<Room>,
|
rtc_source: NativeAudioSource,
|
track: LocalAudioTrack,
|
}
|
|
impl BotAudioOutputSink {
|
async fn publish(
|
room: Arc<Room>,
|
room_name: &str,
|
participant_identity: &str,
|
track_name: &str,
|
sample_rate: u32,
|
num_channels: u32,
|
) -> Result<Self> {
|
let rtc_source = NativeAudioSource::new(
|
AudioSourceOptions::default(),
|
sample_rate,
|
num_channels,
|
1000,
|
);
|
let track = LocalAudioTrack::create_audio_track(
|
track_name,
|
RtcAudioSource::Native(rtc_source.clone()),
|
);
|
|
room.local_participant()
|
.publish_track(
|
LocalTrack::Audio(track.clone()),
|
TrackPublishOptions::default(),
|
)
|
.await
|
.map_err(|error| {
|
anyhow!(
|
"failed to publish bot audio track in room {room_name} for participant {participant_identity}: {error}"
|
)
|
})?;
|
|
info!(
|
room_id = %room_name,
|
participant_alias = %redact(participant_identity),
|
track_name = %track_name,
|
sample_rate,
|
num_channels,
|
"runtime helper published bot audio track"
|
);
|
|
Ok(Self {
|
room,
|
rtc_source,
|
track,
|
})
|
}
|
|
async fn write_pcm_frame(&self, frame: &audio::PcmFrame) -> Result<()> {
|
let audio_frame = AudioFrame {
|
data: frame.data.as_slice().into(),
|
sample_rate: frame.sample_rate,
|
num_channels: frame.num_channels,
|
samples_per_channel: frame.samples_per_channel,
|
};
|
self.rtc_source
|
.capture_frame(&audio_frame)
|
.await
|
.map_err(|error| {
|
anyhow!("failed to capture pcm frame into livekit audio source: {error}")
|
})
|
}
|
|
fn clear_buffer(&self) {
|
self.rtc_source.clear_buffer();
|
}
|
|
async fn close(&self) -> Result<()> {
|
self.room
|
.local_participant()
|
.unpublish_track(&self.track.sid())
|
.await
|
.map(|_| ())
|
.map_err(|error| {
|
anyhow!(
|
"failed to unpublish bot audio track {}: {error}",
|
self.track.sid()
|
)
|
})
|
}
|
}
|
|
async fn wait_for_shutdown_signal() -> Result<()> {
|
#[cfg(unix)]
|
{
|
use tokio::signal::unix::{SignalKind, signal};
|
let mut terminate =
|
signal(SignalKind::terminate()).context("failed to listen for SIGTERM")?;
|
tokio::select! {
|
result = tokio::signal::ctrl_c() => {
|
result.context("failed to listen for ctrl-c")?;
|
}
|
_ = terminate.recv() => {}
|
}
|
return Ok(());
|
}
|
|
#[cfg(not(unix))]
|
{
|
tokio::signal::ctrl_c()
|
.await
|
.context("failed to listen for ctrl-c")?;
|
Ok(())
|
}
|
}
|
