Compare commits

...
17 Commits
Author SHA1 Message Date
tdfischer bcf37182fe fix audio playback of 8Khz samples, further cut down on clones/copies 2026-06-22 16:18:52 +02:00
tdfischer 2d95d0c6aa clippy++ 2026-06-22 16:18:13 +02:00
tdfischer b7559d7fb6 scene: conversation: use Display instead of ToString for clippy++ 2026-06-22 16:17:49 +02:00
tdfischer 74a823d1c2 audio: implement random SFX output 2026-06-22 13:54:42 +02:00
tdfischer 85fa485833 main: run ratatui handler on crash 2026-06-22 13:53:42 +02:00
tdfischer 7563ab85ca clippy++ and report internal error types from tools 2026-06-22 13:53:33 +02:00
tdfischer 242771332c prediction: make mixx loading a little more efficient 2026-06-22 10:47:03 +02:00
tdfischer 7cbff539df src: drop a lot of unwraps() 2026-06-22 10:23:24 +02:00
tdfischer 70ec40a880 prediction: rewrite the prompting stack to create the ship computer as a second character, and let characters/agents operate autonomously with their own tasks and event queues 2026-06-22 08:57:49 +02:00
tdfischer 2d7153eaf7 artifacts: fix dedupe bug 2026-06-22 08:55:59 +02:00
tdfischer 99323c9683 gitignore: ignore mixxx database and out.log 2026-06-22 08:52:25 +02:00
tdfischer 89125d2def prediction: split out maintenance (and thereby the logging interface) of the UI conversation to a separate task, so log::* can work in realtime. 2026-06-17 22:16:19 +02:00
tdfischer a8a44dae63 prediction: push the bulk of the main event loop into the session impl finally 2026-06-17 20:12:58 +02:00
tdfischer cbf7cbd1dd audio: rewrite the audio stack with a more modular architecture and less code 2026-06-17 20:00:01 +02:00
tdfischer e78a2c3215 musicbrainz: dedupe some code 2026-06-17 12:04:26 +02:00
tdfischer 8716350a4e beets: make beets command async 2026-06-17 12:04:00 +02:00
tdfischer 3a8130d785 artifacts: rewrite the entire artifact querying layer to create modular 'tools' and 'datasource's 2026-06-17 11:09:50 +02:00
26 changed files with 2380 additions and 898 deletions
+2
View File
@@ -2,3 +2,5 @@
save.json
.env
audio.json
mixxxdb.sqlite*
out.log
Generated
+231 -3
View File
@@ -492,6 +492,17 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "chacha20"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
dependencies = [
"cfg-if",
"cpufeatures 0.3.0",
"rand_core 0.10.1",
]
[[package]]
name = "chrono"
version = "0.4.44"
@@ -693,6 +704,15 @@ dependencies = [
"libc",
]
[[package]]
name = "cpufeatures"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
dependencies = [
"libc",
]
[[package]]
name = "crc32fast"
version = "1.5.0"
@@ -1198,6 +1218,7 @@ dependencies = [
"minify",
"musicbrainz_rs",
"oximedia-metering",
"rand 0.10.1",
"ratatui",
"rc-writer",
"rdf-types",
@@ -1210,12 +1231,15 @@ dependencies = [
"sqlite",
"static-iref",
"static_cell",
"symphonia",
"tempfile",
"textwrap",
"throbber-widgets-tui",
"tokio",
"tokio-stream",
"tui-input",
"tui-skeleton",
"uuid",
]
[[package]]
@@ -1250,6 +1274,12 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "extended"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "af9673d8203fcb076b19dfd17e38b3d4ae9f44959416ea532ce72415a6020365"
[[package]]
name = "eyre"
version = "0.6.12"
@@ -1522,6 +1552,7 @@ dependencies = [
"cfg-if",
"libc",
"r-efi 6.0.0",
"rand_core 0.10.1",
"wasip2",
"wasip3",
]
@@ -3416,6 +3447,17 @@ dependencies = [
"rand_core 0.9.5",
]
[[package]]
name = "rand"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207"
dependencies = [
"chacha20",
"getrandom 0.4.2",
"rand_core 0.10.1",
]
[[package]]
name = "rand_chacha"
version = "0.9.0"
@@ -3441,6 +3483,12 @@ dependencies = [
"getrandom 0.3.4",
]
[[package]]
name = "rand_core"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
[[package]]
name = "rand_xoshiro"
version = "0.6.0"
@@ -3652,6 +3700,12 @@ dependencies = [
"regex-syntax",
]
[[package]]
name = "regex-lite"
version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973"
[[package]]
name = "regex-syntax"
version = "0.8.10"
@@ -4163,7 +4217,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
dependencies = [
"cfg-if",
"cpufeatures",
"cpufeatures 0.2.17",
"digest",
]
@@ -4524,6 +4578,179 @@ version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "symphonia"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1758d6c853020a7244de03cc3e0185eaea3f58715122422dd3cc7452e6d4c16a"
dependencies = [
"lazy_static",
"symphonia-bundle-flac",
"symphonia-bundle-mp3",
"symphonia-codec-aac",
"symphonia-codec-adpcm",
"symphonia-codec-alac",
"symphonia-codec-pcm",
"symphonia-codec-vorbis",
"symphonia-core",
"symphonia-format-mkv",
"symphonia-format-ogg",
"symphonia-format-riff",
"symphonia-metadata",
]
[[package]]
name = "symphonia-bundle-flac"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee69ad01236a67260b82fd1ff9790dd75ead29f2f46af145e63b7e72273e0e03"
dependencies = [
"log",
"symphonia-common",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-bundle-mp3"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "350f1f2f2e19ad4dd315db94304d1eb361b29af070681f94e51b8fdaad769546"
dependencies = [
"lazy_static",
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-aac"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1979c515a76371b186aad2feff5f23e21cbec775bf95de08bf1e3af92a2ad76"
dependencies = [
"lazy_static",
"log",
"symphonia-common",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-adpcm"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ebbdfd76d6cc5a601c6292a44357c5b7c82f2cd7cdc0f171421f5c5cff0ea1f"
dependencies = [
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-alac"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a149cbfc7fb5c405d123a273227d31de17138419552112bf1aa7b73e65827b8"
dependencies = [
"log",
"symphonia-common",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-pcm"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50baee168f0e9dcf6ba7fc06e8b57eb62072a4490cc7cf13af77e72baae5d328"
dependencies = [
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-vorbis"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "45b07b4423cd8e0fc472575909a5554b12c2f58e3c190b38c24f042e732fd8de"
dependencies = [
"log",
"symphonia-common",
"symphonia-core",
]
[[package]]
name = "symphonia-common"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8257891ffa7f05e02b58f4761e2abf7e5278c8744fd59e981559e050f86eef55"
dependencies = [
"log",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-core"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "95ec293b5f288383b72a7bffcade6b2860b642cf66f28b3bd5967349a49938b1"
dependencies = [
"bitflags 2.11.1",
"bytemuck",
"lazy_static",
"log",
"num-complex",
"rustfft",
"smallvec",
]
[[package]]
name = "symphonia-format-mkv"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb17713e134f5ad316c2690fa3104590ccc85842cdbcf82c3cd1a845cb08aa74"
dependencies = [
"lazy_static",
"log",
"symphonia-common",
"symphonia-core",
]
[[package]]
name = "symphonia-format-ogg"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b05a67e02b1e4fca1a261ba4fe06910a9357489ad8c36aafdd2960e9c6559433"
dependencies = [
"log",
"symphonia-common",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-format-riff"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "17424452a777666d3eaf09a5c651029b15b6a333812fcc5b5474f2a3f0cff3f0"
dependencies = [
"extended",
"log",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-metadata"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a31acf5cd623398a6208e2225d18f4b20f761c55098a796a5247ad516a4a8681"
dependencies = [
"lazy_static",
"log",
"regex-lite",
"smallvec",
"symphonia-core",
]
[[package]]
name = "syn"
version = "1.0.109"
@@ -5190,13 +5417,14 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "uuid"
version = "1.23.1"
version = "1.23.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76"
checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7"
dependencies = [
"atomic",
"getrandom 0.4.2",
"js-sys",
"serde_core",
"wasm-bindgen",
]
+4
View File
@@ -22,6 +22,7 @@ log = "0.4.32"
minify = "1.3.0"
musicbrainz_rs = { version = "0.13.0", features = ["async"] }
oximedia-metering = "0.1.7"
rand = "0.10.1"
ratatui = "0.30.0"
rc-writer = "1.1.10"
rdf-types = "0.22.5"
@@ -34,9 +35,12 @@ serde_json = "1.0.150"
sqlite = "0.37.0"
static-iref = "3.0.0"
static_cell = "2.1.1"
symphonia = { version = "0.6.0", features = ["all-codecs"] }
tempfile = "3.27.0"
textwrap = "0.16.2"
throbber-widgets-tui = "0.11.0"
tokio = { version = "1.52.3", features = ["full"] }
tokio-stream = "0.1.18"
tui-input = "0.15.3"
tui-skeleton = "0.3.0"
uuid = { version = "1.23.3", features = ["serde", "v4"] }
+1
View File
@@ -1,3 +1,4 @@
fn main() {
println!("cargo::rerun-if-changed=src/system-prompt.txt");
println!("cargo::rerun-if-changed=src/computer-prompt.txt");
}
+190
View File
@@ -0,0 +1,190 @@
use std::{collections::HashMap, ops::{Deref, DerefMut}};
use std::fmt::Debug;
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::artifacts::{Artifact, Contents, Merge, SourceID, beets::BeetsDB, mixxx::MixxxDB, musicbrainz::MBQuery, tools::DataSource};
pub struct ArtifactRef<'a> {
id: Uuid,
archive: &'a Archive
}
pub struct ArtifactRefMut<'a> {
id: Uuid,
archive: &'a mut Archive
}
impl<'a> ArtifactRefMut<'a> {
pub fn downgrade(self) -> ArtifactRef<'a> {
ArtifactRef { id: self.id, archive: self.archive }
}
}
impl<'a> Deref for ArtifactRef<'a> {
type Target = Artifact;
fn deref(&self) -> &Self::Target {
self.archive.contents.get(&self.id).unwrap()
}
}
impl<'a> Deref for ArtifactRefMut<'a> {
type Target = Artifact;
fn deref(&self) -> &Self::Target {
self.archive.contents.get(&self.id).unwrap()
}
}
impl<'a> DerefMut for ArtifactRefMut<'a> {
fn deref_mut(&mut self) -> &mut Self::Target {
self.archive.contents.get_mut(&self.id).unwrap()
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Archive {
#[serde(flatten)]
contents: HashMap<Uuid, Artifact>
}
impl Archive {
pub fn len(&self) -> usize {
self.contents.len()
}
// track, album, artist
pub fn stats(&self) -> (usize, usize, usize) {
self.contents.values().map(|artifact| {
match artifact.contents() {
Contents::Track(_) => (1, 0, 0),
Contents::Album(_) => (0, 1, 0),
Contents::Artist(_) => (0, 0, 1),
}
}).reduce(|acc, e| {
(acc.0 + e.0, acc.1 + e.1, acc.2 + e.2)
}).unwrap_or_default()
}
pub fn get<'a>(&'a self, id: &Uuid) -> Option<ArtifactRef<'a>> {
if self.contents.contains_key(id) {
Some(ArtifactRef { id: *id, archive: self })
} else {
None
}
}
pub fn get_mut<'a>(&'a mut self, id: &Uuid) -> Option<ArtifactRefMut<'a>> {
if self.contents.contains_key(id) {
Some(ArtifactRefMut { id: *id, archive: self })
} else {
None
}
}
pub async fn data_sync<Src: DataSource>(&mut self, datasrc: &mut Src, source: SourceID) -> usize where Src::Error: Debug {
let mut count = 0;
let pending = self.contents.iter_mut().filter_map(|(_, artifact)| {
if !artifact.sources.contains(&source) {
Some(artifact)
} else {
None
}
});
let futures = futures::stream::FuturesUnordered::new();
for artifact in pending {
futures.push(datasrc.synchronize(artifact));
}
let results: Vec<_> = futures.collect().await;
for result in results {
match result {
Ok(new_pending) => {
count += new_pending.len() + 1;
for new in new_pending {
self.insert(new);
}
},
Err(err) => {
log::error!("Failed to synchronize: {:?}", err);
}
}
}
count
}
pub async fn synchronize(&mut self) -> usize {
log::debug!("Synchronizing records");
let mut count = 0;
log::debug!("Synchronizing Mixxx");
count += self.data_sync(&mut MixxxDB, SourceID::Mixxx).await;
log::debug!("Synchronizing Beets");
count += self.data_sync(&mut BeetsDB, SourceID::Beets).await;
log::debug!("Synchronizing Musicbrainz");
count += self.data_sync(&mut MBQuery, SourceID::Musicbrainz).await;
log::debug!("Updated {} records", count);
count
}
pub fn insert<'a>(&'a mut self, artifact: Artifact) -> ArtifactRef<'a> {
// If we are inserting a new artifact with a complete MBID...
if let Some(mbid) = artifact.mbid {
let search_id = mbid;
// If an entry already exists keyed by this MBID, merge into it
if let Some(existing) = self.contents.get_mut(&search_id) {
existing.merge(artifact);
ArtifactRef { id: search_id, archive: self }
} else {
// Otherwise, attempt to find existing artifacts with the same contents (but no MBID)
let mut targets: Vec<(Uuid, Artifact)> = self.contents.extract_if(|_, v| { v.contents == artifact.contents }).collect();
if let Some((_target_id, mut target)) = targets.pop() {
// Merge any other extracted targets into the primary one
for (_, next) in targets {
target.merge(next);
}
// Merge the incoming artifact into the merged target
target.merge(artifact);
// Insert merged target under the canonical MBID key
self.contents.insert(search_id, target);
ArtifactRef { id: search_id, archive: self }
} else {
// No matching content found: insert under the MBID key
self.contents.insert(search_id, artifact);
ArtifactRef { id: search_id, archive: self }
}
}
} else {
// Otherwise, we attempt to merge it in. In the end, there will somehow still be a record with this mbid
let mut targets: Vec<(Uuid, Artifact)> = self.contents.extract_if(|_, v| { v.contents == artifact.contents }).collect();
if let Some((target_id, mut target)) = targets.pop() {
let next_id = if let Some(mbid) = artifact.mbid {
// If the new artifact has an mbid, we start using that as the archive key
mbid
} else {
// Otherwise, why regenerate a new one?
target_id
};
for (_, next) in targets {
target.merge(next);
}
target.merge(artifact);
// Re-insert the merged target back into the archive under the chosen id
self.contents.insert(next_id, target);
ArtifactRef { id: next_id, archive: self }
} else {
let new_id = Uuid::new_v4();
self.contents.insert(new_id, artifact);
ArtifactRef { id: new_id, archive: self }
}
}
}
}
+69 -16
View File
@@ -1,30 +1,83 @@
use std::collections::HashSet;
use bandcamp::SearchResultItem;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use crate::artifacts::{Album, Artifact, Artist, SourceID};
use crate::artifacts::{Album, Artifact, ArtifactBuilder, Artist, SourceID, Track, tools::{DataSource, ToolDescription}};
#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
pub struct BandcampQueryArgs {
pub query: String
}
impl Into<Artifact> for bandcamp::Artist {
fn into(self) -> Artifact {
Artifact::Artist(Artist { name: self.name, bio: self.bio, location: self.location, sources: HashSet::from([SourceID::Bandcamp(self.id)])})
impl From<bandcamp::Artist> for Artifact {
fn from(val: bandcamp::Artist) -> Self {
ArtifactBuilder::new(SourceID::Bandcamp).contents(Artist { name: val.name, bio: val.bio, location: val.location }).build()
}
}
impl Into<Artifact> for bandcamp::Album {
fn into(self) -> Artifact {
Artifact::Album(Album {
about: self.about,
title: self.title,
artist: self.band.name,
credits: self.credits,
release_date: Some(self.release_date),
sources: HashSet::from([SourceID::Bandcamp(self.id)])
})
impl From<bandcamp::Album> for Artifact {
fn from(val: bandcamp::Album) -> Self {
ArtifactBuilder::new(SourceID::Bandcamp)
.contents(Album {
about: val.about,
title: val.title,
artist: val.band.name,
credits: val.credits,
release_date: Some(val.release_date)
}).build()
}
}
pub struct BandcampSource;
impl DataSource for BandcampSource {
type Args = BandcampQueryArgs;
type Error = bandcamp::Error;
async fn synchronize(&self, _artifact: &mut Artifact) -> Result<Vec<Artifact>, Self::Error> {
todo!()
}
async fn query(&self, args: &Self::Args) -> Result<Vec<Artifact>, Self::Error> {
log::debug!("Fetching artifacts from Bandcamp with {:?}", args);
let mut json_results = vec![];
if let Ok(results) = bandcamp::search(args.query.as_str()).await {
for result in results {
log::debug!("Result: {:?}", result);
match result {
SearchResultItem::Artist(data) => {
// TODO: The artist and album detailed fetchers should also be separate args
let result = bandcamp::fetch_artist(data.artist_id).await?.into();
json_results.push(result);
},
SearchResultItem::Album(data) => {
let result = bandcamp::fetch_album(data.band_id, data.album_id).await?.into();
json_results.push(result);
},
SearchResultItem::Track(data) => {
let result = ArtifactBuilder::new(SourceID::Bandcamp)
.contents(Track {
title: data.name,
artist: Some(data.band_name),
album: data.album_name,
..Default::default()
}).build();
json_results.push(result);
}
_ => ()
}
}
}
Ok(json_results)
}
}
impl ToolDescription for BandcampSource {
fn description(&self) -> &str {
"Scans Bandcamp to find artifacts to use in the scene that match the given search parameters. To find an artist, provide only the artist name. To find an album, provide the artist and the album."
}
fn name(&self) -> &str {
"query_bandcamp"
}
}
+121 -41
View File
@@ -1,18 +1,27 @@
use std::{collections::HashSet, process::{Command, Stdio}};
use std::process::Stdio;
use tokio::process::Command;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::artifacts::{Artifact, SourceID, Track};
use crate::artifacts::{Artifact, ArtifactBuilder, Contents, Merge, SourceID, Track, tools::{DataSource, ToolDescription}};
#[derive(Debug, Default, Serialize, Deserialize, Clone, JsonSchema)]
pub struct BeatsQueryArgs {
artist: Option<String>,
album: Option<String>,
genre: Option<String>,
title: Option<String>,
year: Option<u32>
pub artist: Option<String>,
pub album: Option<String>,
pub genre: Option<String>,
pub title: Option<String>,
pub year: Option<u32>,
pub label: Option<String>
}
#[derive(Debug, Default, Serialize, Deserialize, Clone, JsonSchema)]
pub struct BeatsQueryMultiArgs {
args: Vec<BeatsQueryArgs>
}
#[derive(Debug, Default, Deserialize)]
@@ -26,69 +35,140 @@ struct BeetsTrack {
mb_trackid: Option<String>
}
impl Into<Artifact> for BeetsTrack {
fn into(self) -> Artifact {
let sources = if let Some(mbid) = self.mb_trackid {
HashSet::from([SourceID::Beets, SourceID::Musicbrainz(mbid)])
} else {
HashSet::from([SourceID::Beets])
};
Artifact::Track(Track {
title: self.title,
label: self.label,
year: Some(self.year),
genres: self.genres.unwrap_or_default(),
album: Some(self.album),
artist: Some(self.artist),
impl From<BeetsTrack> for Artifact {
fn from(val: BeetsTrack) -> Self {
let track_data = Track {
title: val.title,
label: val.label,
year: Some(val.year),
genres: val.genres.unwrap_or_default(),
album: Some(val.album),
artist: Some(val.artist),
bpm: None,
sources
})
};
let builder = ArtifactBuilder::new(SourceID::Beets)
.contents(track_data);
if let Ok(mbid) = Uuid::parse_str(&val.mb_trackid.unwrap_or_default()) {
builder.mbid(mbid).build()
} else {
builder.build()
}
}
}
impl BeatsQueryArgs {
pub fn execute(self) -> Result<Vec<Artifact>, ()> {
#[derive(Debug)]
#[allow(unused)]
pub enum BeetsError {
Command(std::io::Error),
Json(serde_json::Error),
EmptyQuery
}
impl From<std::io::Error> for BeetsError {
fn from(value: std::io::Error) -> Self {
Self::Command(value)
}
}
impl From<serde_json::Error> for BeetsError {
fn from(value: serde_json::Error) -> Self {
Self::Json(value)
}
}
pub struct BeetsDB;
impl BeetsDB {
async fn query_multi(&self, args: &BeatsQueryMultiArgs) -> Result<Vec<Artifact>, BeetsError> {
let mut ret = vec![];
for arg in &args.args {
for artifact in self.query_single(arg).await.unwrap_or_default() {
ret.push(artifact);
}
}
Ok(ret)
}
async fn query_single(&self, args: &BeatsQueryArgs) -> Result<Vec<Artifact>, BeetsError> {
let mut beets_cmd = Command::new("beet");
beets_cmd.args(["export", "-f", "json", "-i", "title,label,year,genres,album,artist,mb_trackid"]);
let mut valid = false;
if let Some(artist) = self.artist {
if let Some(ref artist) = args.artist {
beets_cmd.arg(format!("artist:{}", artist));
valid = true;
}
if let Some(genre) = self.genre {
if let Some(ref genre) = args.genre {
beets_cmd.arg(format!("genre:{}", genre));
valid = true;
}
if let Some(album) = self.album {
if let Some(ref album) = args.album {
beets_cmd.arg(format!("album:{}", album));
valid = true;
}
if let Some(title) = self.title {
if let Some(ref title) = args.title {
beets_cmd.arg(format!("title:{}", title));
valid = true;
}
if let Some(year) = self.year {
if let Some(year) = args.year {
beets_cmd.arg(format!("year:{}", year));
valid = true;
}
if let Some(ref label) = args.label {
beets_cmd.arg(format!("label:{}", label));
valid = true;
}
if !valid {
log::warn!("Tried to execute an empty beets query");
return Err(())
return Err(BeetsError::EmptyQuery)
}
log::debug!("Executing beets: {:?}", beets_cmd);
if let Ok(output) = beets_cmd.stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap().wait_with_output() {
match serde_json::from_str::<Vec<BeetsTrack>>(str::from_utf8(&output.stdout).unwrap()) {
Ok(track) => Ok(track.into_iter().map(|t| { t.into()}).collect()),
Err(err) => {
log::error!("Failed to decode beets json: {:?}", err);
Err(())
}
let output = beets_cmd.stdout(Stdio::piped()).stderr(Stdio::null()).spawn()?.wait_with_output().await?;
let track = serde_json::from_str::<Vec<BeetsTrack>>(str::from_utf8(&output.stdout).unwrap())?;
Ok(track.into_iter().map(|t| { t.into()}).collect())
}
}
impl DataSource for BeetsDB {
type Args = BeatsQueryMultiArgs;
type Error = BeetsError;
async fn synchronize(&self, artifact: &mut Artifact) -> Result<Vec<Artifact>, Self::Error> {
if let Contents::Track(ref mut target_track) = artifact.contents {
let args = BeatsQueryArgs {
title: Some(target_track.title.clone()),
artist: target_track.artist.clone(),
album: target_track.album.clone(),
..Default::default()
};
let results = self.query(&BeatsQueryMultiArgs { args: vec![args] }).await?;
if let Some(first) = results.first() {
artifact.merge(first.clone());
} else {
log::debug!("Beets could not find {:?}", target_track);
}
} else {
log::error!("Unable to execute query!");
Err(())
}
Ok(vec![])
}
fn query(&self, args: &Self::Args) -> impl Future<Output = Result<Vec<Artifact>, Self::Error>> {
self.query_multi(args)
}
}
impl ToolDescription for BeetsDB {
fn description(&self) -> &str {
"Queries the ship's musical artifact archives for tracks matching the given search parameters"
}
fn name(&self) -> &str {
"query_beets"
}
}
+48 -25
View File
@@ -1,8 +1,8 @@
use std::collections::HashSet;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use sqlite::OpenFlags;
use crate::artifacts::{Album, Artifact, Artist, SourceID, Track};
use crate::artifacts::{Album, Artifact, ArtifactBuilder, Artist, SourceID, Track, tools::{DataSource, ToolDescription}};
#[derive(Debug)]
#[allow(unused)]
@@ -16,11 +16,24 @@ impl From<sqlite::Error> for MixxxError {
}
}
pub struct MixxxDB(());
pub struct MixxxDB;
impl MixxxDB {
pub fn load(playlist_name: &str) -> Result<Vec<Artifact>, MixxxError> {
#[derive(Serialize, Deserialize, Debug, Default, JsonSchema)]
pub struct MixxxQuery {
pub playlist_name: String
}
impl DataSource for MixxxDB {
type Args = MixxxQuery;
type Error = MixxxError;
async fn synchronize(&self, _artifact: &mut Artifact) -> Result<Vec<Artifact>, Self::Error> {
Ok(vec![])
}
async fn query(&self, args: &Self::Args) -> Result<Vec<Artifact>, Self::Error> {
let mut ret = vec![];
let playlist_name = args.playlist_name.as_str();
log::info!("Loading Mixxx playlist {}", playlist_name);
let connection = sqlite::Connection::open_thread_safe_with_flags("mixxxdb.sqlite", OpenFlags::new().with_read_only())?;
let query = "SELECT id FROM Playlists WHERE name = ? ORDER BY id DESC LIMIT 1";
@@ -36,29 +49,39 @@ impl MixxxDB {
let artist = track.try_read::<&str, _>("artist").unwrap_or("Unknown Artist");
let album = track.try_read::<&str, _>("album").unwrap_or("Unknown Album");
let bpm = track.try_read::<f64, _>("bpm").unwrap_or(0.);
ret.push(Artifact::Track(Track {
artist: Some(artist.into()),
album: Some(album.into()),
title: title.into(),
bpm: Some(bpm),
sources: HashSet::from([SourceID::Mixxx]),
..Default::default()
}));
ret.push(ArtifactBuilder::new(SourceID::Mixxx)
.contents(Track {
artist: Some(artist.into()),
album: Some(album.into()),
title: title.into(),
bpm: Some(bpm),
..Default::default()
}).build());
ret.push(Artifact::Album(Album {
artist: artist.into(),
title: album.into(),
sources: HashSet::from([SourceID::Mixxx]),
..Default::default()
}));
ret.push(ArtifactBuilder::new(SourceID::Mixxx)
.contents(Album {
artist: artist.into(),
title: album.into(),
..Default::default()
}).build());
ret.push(Artifact::Artist(Artist {
name: artist.into(),
sources: HashSet::from([SourceID::Mixxx]),
..Default::default()
}));
ret.push(ArtifactBuilder::new(SourceID::Mixxx)
.contents(Artist {
name: artist.into(),
..Default::default()
}).build());
}
Ok(ret)
}
}
impl ToolDescription for MixxxDB {
fn description(&self) -> &str {
"Loads artifacts from a given Mixxx playlist name"
}
fn name(&self) -> &str {
"query_mixxx"
}
}
+150 -54
View File
@@ -1,17 +1,21 @@
use std::collections::{HashMap, HashSet};
use std::{collections::HashSet, fmt::Debug, };
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
pub mod bandcamp;
pub mod mixxx;
pub mod beets;
pub mod musicbrainz;
pub mod archive;
pub mod tools;
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum SourceID {
Bandcamp(u64),
Musicbrainz(String),
Bandcamp,
Musicbrainz,
Mixxx,
Beets
}
@@ -23,8 +27,6 @@ pub struct Artist {
pub bio: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub location: Option<String>,
pub sources: HashSet<SourceID>
}
impl PartialEq for Artist {
@@ -35,13 +37,9 @@ impl PartialEq for Artist {
true
}
fn ne(&self, other: &Self) -> bool {
!self.eq(other)
}
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default)]
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct Album {
pub title: String,
pub artist: String,
@@ -50,9 +48,17 @@ pub struct Album {
#[serde(skip_serializing_if = "Option::is_none")]
pub credits: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub release_date: Option<DateTime<Utc>>,
pub release_date: Option<DateTime<Utc>>
}
pub sources: HashSet<SourceID>
impl PartialEq for Album {
fn eq(&self, other: &Self) -> bool {
if self.title != other.title || self.artist != other.artist {
return false;
}
true
}
}
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
@@ -70,9 +76,7 @@ pub struct Track {
#[serde(skip_serializing_if = "Option::is_none")]
pub artist: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub bpm: Option<f64>,
pub sources: HashSet<SourceID>
pub bpm: Option<f64>
}
impl PartialEq for Track {
@@ -81,64 +85,41 @@ impl PartialEq for Track {
return false;
}
if self.artist.is_some() && self.artist != other.artist {
if other.artist.is_some() && self.artist.is_some() && self.artist != other.artist {
return false;
}
if self.album.is_some() && self.album != other.album {
if other.album.is_some() && self.album.is_some() && self.album != other.album {
return false;
}
true
}
fn ne(&self, other: &Self) -> bool {
!self.eq(other)
}
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
pub enum Artifact {
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
#[serde(tag = "type")]
pub enum Contents {
Artist(Artist),
Album(Album),
Track(Track)
}
macro_rules! merge_fields {
($this:expr, $that:expr, $field:tt) => {
if $this.$field.is_none() {
$this.$field = $that.$field;
}
};
($this:tt, $that:tt, $($fields:tt),+) => {
$(
merge_fields!($this, $that, $fields);
for src in &$that.sources {
$this.sources.insert(src.clone());
}
)+
impl From<Artist> for Contents {
fn from(value: Artist) -> Self {
Self::Artist(value)
}
}
impl Merge for Artifact {
fn merge(&mut self, other: Self) {
if *self != other {
return;
}
impl From<Album> for Contents {
fn from(value: Album) -> Self {
Self::Album(value)
}
}
match (self, other) {
(Self::Track(this_track), Self::Track(that_track)) => {
merge_fields!(this_track, that_track, album, label, year, artist, bpm);
},
(Self::Album(this_album), Self::Album(that_album)) => {
merge_fields!(this_album, that_album, about, credits, release_date);
},
(Self::Artist(this_artist), Self::Artist(that_artist)) => {
merge_fields!(this_artist, that_artist, bio, location);
},
_ => ()
}
impl From<Track> for Contents {
fn from(value: Track) -> Self {
Self::Track(value)
}
}
@@ -156,4 +137,119 @@ impl<M: Merge + PartialEq> Merge for Vec<M> {
pub trait Merge {
fn merge(&mut self, other: Self);
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
pub struct Artifact {
#[serde(skip_serializing_if = "Option::is_none")]
mbid: Option<Uuid>,
#[serde(flatten)]
contents: Contents,
sources: HashSet<SourceID>,
}
#[derive(Debug)]
pub struct ArtifactBuilder {
contents: Option<Contents>,
mbid: Option<Uuid>,
source: SourceID,
}
impl ArtifactBuilder {
pub fn new(source: SourceID) -> Self {
Self {
contents: None,
mbid: None,
source,
}
}
pub fn contents<T: Into<Contents>>(mut self, contents: T) -> Self {
self.contents = Some(contents.into());
self
}
pub fn mbid<T: Into<Uuid>>(mut self, mbid: T) -> Self {
self.mbid = Some(mbid.into());
self
}
pub fn build(self) -> Artifact {
Artifact {
mbid: self.mbid,
contents: self.contents.unwrap(),
sources: HashSet::from_iter([self.source]),
}
}
}
impl Artifact {
pub fn contents(&self) -> &Contents {
&self.contents
}
}
impl Merge for Artifact {
fn merge(&mut self, other: Self) {
self.contents.merge(other.contents);
if self.mbid.is_none() {
self.mbid = other.mbid;
}
for src in other.sources {
self.sources.insert(src);
}
}
}
macro_rules! merge_fields {
($this:expr, $that:expr, $field:tt) => {
if $this.$field.is_none() {
$this.$field = $that.$field;
}
};
($this:tt, $that:tt, $($fields:tt),+) => {
$(
merge_fields!($this, $that, $fields);
)+
}
}
impl Merge for Contents {
fn merge(&mut self, other: Self) {
if *self != other {
return;
}
match (self, other) {
(Self::Track(this_track), Self::Track(that_track)) => {
this_track.merge(that_track);
},
(Self::Album(this_album), Self::Album(that_album)) => {
this_album.merge(that_album);
},
(Self::Artist(this_artist), Self::Artist(that_artist)) => {
this_artist.merge(that_artist);
},
_ => ()
}
}
}
impl Merge for Track {
fn merge(&mut self, other: Self) {
merge_fields!(self, other, album, label, year, artist, bpm);
}
}
impl Merge for Artist {
fn merge(&mut self, other: Self) {
merge_fields!(self, other, bio, location);
}
}
impl Merge for Album {
fn merge(&mut self, other: Self) {
merge_fields!(self, other, about, credits, release_date);
}
}
+136 -36
View File
@@ -1,50 +1,150 @@
use std::collections::HashSet;
use musicbrainz_rs::entity::recording::Recording;
use musicbrainz_rs::entity::artist_credit::ArtistCredit;
use musicbrainz_rs::entity::release::Release;
use musicbrainz_rs::{ApiEndpointError, entity::recording::Recording};
use musicbrainz_rs::prelude::*;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::artifacts::{Album, Artifact, Artist, SourceID, Track};
use crate::artifacts::tools::{DataSource, ToolDescription};
use crate::artifacts::{Album, Artifact, ArtifactBuilder, Artist, Contents, Merge, SourceID, Track};
#[derive(Debug, Default, Deserialize, Serialize, JsonSchema)]
pub struct MusicbrainzQueryArgs {
pub mb_ids: Vec<String>
impl From<Recording> for Track {
fn from(value: Recording) -> Self {
let artist = value.artist_credit.unwrap_or_default().first().map(|x| x.name.clone() );
let album = value.releases.unwrap_or_default().first().map(|x| x.title.clone() );
Self {
title: value.title,
artist,
album,
..Default::default()
}
}
}
pub async fn search_artifacts(query: MusicbrainzQueryArgs) -> Result<Vec<Artifact>, musicbrainz_rs::ApiEndpointError> {
let mut ret = vec![];
for mbid in query.mb_ids {
let track = Recording::fetch()
.id(&mbid)
.with_releases().with_artists().with_annotations().execute_async().await?;
impl From<Release> for Album {
fn from(value: Release) -> Self {
let artist = value.artist_credit.unwrap_or_default().first().map(|x| x.name.clone() );
Self {
about: value.annotation,
title: value.title,
artist: artist.unwrap_or_default(),
..Default::default()
}
}
}
impl From<ArtistCredit> for Artist {
fn from(value: ArtistCredit) -> Self {
Self {
bio: value.artist.annotation,
location: value.artist.country,
name: value.name,
}
}
}
pub struct MBQuery;
impl MBQuery {
fn extract_recording_data(track: Recording) -> (Artifact, Vec<Artifact>) {
let mut ret = vec![];
let ret_track = ArtifactBuilder::new(SourceID::Musicbrainz)
.contents(Track::from(track.clone()))
.mbid(Uuid::parse_str(&track.id).unwrap())
.build();
for release in track.releases.unwrap_or_default() {
log::debug!("Found new release: {:?}", release);
let first_artist = release.artist_credit.unwrap_or_default().first().unwrap().clone();
ret.push(Artifact::Album(Album {
title: release.title.clone(),
artist: first_artist.name.clone(),
about: release.annotation,
sources: HashSet::from([SourceID::Musicbrainz(release.id.clone())]),
..Default::default()
}));
ret.push(Artifact::Track(Track {
album: Some(release.title),
title: track.title.clone(),
artist: Some(first_artist.artist.name.clone()),
sources: HashSet::from([SourceID::Musicbrainz(release.id.clone())]),
..Default::default()
}));
ret.push(Artifact::Artist(Artist {
name: first_artist.name,
bio: first_artist.artist.annotation,
location: first_artist.artist.area.and_then(|area| { Some(area.name) }),
sources: HashSet::from([SourceID::Musicbrainz(release.id)]),
..Default::default()
}))
ret.push(ArtifactBuilder::new(SourceID::Musicbrainz)
.mbid(Uuid::parse_str(&release.id).unwrap())
.contents(Album::from(release))
.build());
}
for artist in track.artist_credit.unwrap_or_default() {
ret.push(ArtifactBuilder::new(SourceID::Musicbrainz)
.mbid(Uuid::parse_str(&artist.artist.id).unwrap())
.contents(Artist::from(artist))
.build());
}
(ret_track, ret)
}
}
impl DataSource for MBQuery {
type Error = ApiEndpointError;
type Args = MusicbrainzQueryArgs;
async fn synchronize(&self, artifact: &mut Artifact) -> Result<Vec<Artifact>, Self::Error> {
let mut ret = vec![];
if artifact.mbid.is_none() {
return Ok(ret);
}
let artifact_id = artifact.mbid.unwrap();
log::debug!("Synchronizing {} with musicbrainz", artifact_id);
// FIXME: Need to also synchronize albums and artists
if let Contents::Track(_) = artifact.contents {
let mb_track = Recording::fetch()
.id(&artifact_id.to_string())
.with_releases().with_artists().with_annotations().execute_async().await;
let track = match mb_track {
Ok(track) => track,
Err(err) => {
log::error!("Failed to grab musicbrainz data: {:?}", err);
return Err(err);
}
};
let (track, new_artifacts) = Self::extract_recording_data(track);
ret.push(track.clone());
ret.extend(new_artifacts);
artifact.merge(track);
}
Ok(ret)
}
Ok(ret)
async fn query(&self, args: &Self::Args) -> Result<Vec<Artifact>, Self::Error> {
let mut ret = vec![];
log::debug!("Fetching recording id {}", args.mbid);
let track = Recording::fetch()
.id(&args.mbid)
.with_releases().with_artists().with_annotations().execute_async().await;
let track = match track {
Ok(track) => track,
Err(err) => {
log::error!("Failed to grab musicbrainz data: {:?}", err);
return Err(err)
}
};
let (track, new_artifacts) = Self::extract_recording_data(track);
ret.push(track);
ret.extend(new_artifacts);
Ok(ret)
}
}
impl ToolDescription for MBQuery {
fn description(&self) -> &str {
"Fetches artifacts from Musicbrainz"
}
fn name(&self) -> &str {
"query_musicbrainz"
}
}
#[derive(Debug, Default, Deserialize, Serialize, JsonSchema)]
pub struct MusicbrainzQueryArgs {
pub mbid: String
}
+50
View File
@@ -0,0 +1,50 @@
use async_openai::types::chat::{ChatCompletionTool, ChatCompletionTools, FunctionObjectArgs};
use schemars::{JsonSchema, Schema, schema_for};
use serde::de::DeserializeOwned;
use crate::artifacts::Artifact;
pub trait DataSource: ToolDescription {
type Args: JsonSchema + DeserializeOwned;
type Error;
fn synchronize(&self, artifact: &mut Artifact) -> impl Future<Output = Result<Vec<Artifact>, Self::Error>>;
fn query(&self, args: &Self::Args) -> impl Future<Output = Result<Vec<Artifact>, Self::Error>>;
}
pub trait ToolDescription {
fn description(&self) -> &str;
fn name(&self) -> &str;
}
pub struct Tool {
pub name: String,
pub description: String,
pub schema: Schema
}
impl Tool {
pub fn from_datasource<T: DataSource>(src: &T) -> Self {
Self {
name: src.name().to_string(),
description: src.description().to_string(),
schema: schema_for!(T::Args)
}
}
}
impl From<Tool> for ChatCompletionTool {
fn from(val: Tool) -> Self {
ChatCompletionTool {
function: FunctionObjectArgs::default()
.name(val.name)
.description(val.description)
.parameters(val.schema).build().unwrap()
}
}
}
impl From<Tool> for ChatCompletionTools {
fn from(val: Tool) -> Self {
ChatCompletionTools::Function(val.into())
}
}
+228 -126
View File
@@ -1,8 +1,45 @@
use jack::{AudioIn, AudioOut, ClientOptions, NotificationHandler};
use std::{collections::HashMap, fmt::Display};
use jack::{AudioIn, AudioOut, ClientOptions, NotificationHandler, Port, ProcessScope};
use oximedia_metering::vu_meter::VuMeter;
use serde::{Deserialize, Serialize};
use tokio::sync::*;
use crate::events::AudioRecordRequest;
#[derive(Debug)]
#[allow(unused)]
pub enum AudioError {
Jack(jack::Error),
AudioBufferSend(mpsc::error::SendError<Vec<f32>>),
AudioBufferRecv(mpsc::error::TryRecvError),
AudioRequestSend(watch::error::SendError<AudioRecordRequest>)
}
impl From<jack::Error> for AudioError {
fn from(value: jack::Error) -> Self {
Self::Jack(value)
}
}
impl From<mpsc::error::SendError<Vec<f32>>> for AudioError {
fn from(value: mpsc::error::SendError<Vec<f32>>) -> Self {
Self::AudioBufferSend(value)
}
}
impl From<mpsc::error::TryRecvError> for AudioError {
fn from(value: mpsc::error::TryRecvError) -> Self {
Self::AudioBufferRecv(value)
}
}
impl From<watch::error::SendError<AudioRecordRequest>> for AudioError {
fn from(value: watch::error::SendError<AudioRecordRequest>) -> Self {
Self::AudioRequestSend(value)
}
}
#[derive(Debug)]
pub struct JackClientRef {
killswitch: Option<oneshot::Sender<()>>
@@ -10,7 +47,7 @@ pub struct JackClientRef {
impl Drop for JackClientRef {
fn drop(&mut self) {
self.killswitch.take().unwrap().send(()).unwrap();
self.killswitch.take().expect("Killswitch was already dropped!").send(()).expect("Cannot fire Jack killswitch");
}
}
@@ -21,41 +58,139 @@ pub struct AudioInputControl {
}
impl AudioInputControl {
pub async fn next(&mut self) -> f64 {
self.volume_src.changed().await.unwrap();
*self.volume_src.borrow_and_update()
pub async fn next(&mut self) -> Result<f64, watch::error::RecvError> {
self.volume_src.changed().await?;
Ok(*self.volume_src.borrow_and_update())
}
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Copy, Hash)]
enum Role {
Mic,
Tts,
Sfx
}
impl Display for Role {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let val = match self {
Self::Mic => "Microphone Input",
Self::Tts => "TTS Output",
Self::Sfx => "SFX Output"
};
f.write_str(val)
}
}
impl Role {
fn is_input(&self) -> bool {
matches!(self, Role::Mic)
}
}
#[derive(Debug)]
pub struct MicStream {
pub struct AudioInStream {
pub src: mpsc::Receiver<Vec<f32>>,
pub sample_rate: u32
}
#[derive(Debug)]
pub struct TtsOutStream {
pub struct AudioOutStream {
pub sink: mpsc::Sender<Vec<f32>>,
pub sample_rate: u32
}
struct AudioSource {
port: Port<jack::AudioIn>,
sample_sink: mpsc::Sender<Vec<f32>>,
meter: VuMeter
}
impl AudioSource {
fn new(client: &jack::Client, name: &str) -> Result<(Self, AudioInStream), AudioError> {
let (sample_sink, receiver) = mpsc::channel(32);
let port = client.register_port(name, AudioIn::default())?;
Ok((AudioSource {
port,
sample_sink,
meter: VuMeter::new(client.sample_rate().into(), 1, None)
}, AudioInStream {
sample_rate: client.sample_rate(),
src: receiver
}))
}
fn process(&mut self, scope: &ProcessScope) -> Result<Option<f64>, AudioError> {
if self.port.connected_count()? > 0 {
let buf: Vec<_> = self.port.as_slice(scope).to_vec();
self.meter.process_interleaved(&buf);
self.sample_sink.blocking_send(buf)?;
Ok(self.meter.channel_vu(0))
} else {
Ok(None)
}
}
}
#[derive(Debug)]
pub struct SfxOutStream {
pub sink: mpsc::Sender<Vec<f32>>,
pub sample_rate: u32
struct AudioSink {
output_buf: Vec<f32>,
port: Port<jack::AudioOut>,
sample_src: mpsc::Receiver<Vec<f32>>
}
impl AudioSink {
fn new(client: &jack::Client, name: &str) -> Result<(Self, AudioOutStream), AudioError> {
let (sender, sample_src) = mpsc::channel(32);
let port = client.register_port(name, AudioOut::default())?;
Ok((AudioSink {
output_buf: Vec::with_capacity(1024),
port,
sample_src
}, AudioOutStream {
sample_rate: client.sample_rate(),
sink: sender,
}))
}
fn process(&mut self, scope: &ProcessScope) -> Result<(), AudioError> {
if let Ok(buf) = self.sample_src.try_recv() {
self.output_buf.extend(buf);
}
if self.port.connected_count()? > 0 && !self.output_buf.is_empty() {
let outbuf = self.port.as_mut_slice(scope);
let mut next_segment: Vec<f32> = self.output_buf.drain(..(outbuf.len()).min(self.output_buf.len())).collect();
let underrun = outbuf.len() - next_segment.len();
if underrun > 0 {
log::warn!("Audio stream underrun: {} samples", underrun);
next_segment.extend(std::iter::repeat_n(0., underrun));
}
outbuf.copy_from_slice(&next_segment);
}
Ok(())
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
struct AudioConfig {
mic_in_connections: Vec<String>,
tts_out_connections: Vec<String>,
sfx_out_connections: Vec<String>,
connections: HashMap<Role, Vec<String>>
}
impl AudioConfig {
pub fn load() -> Self {
if let Ok(contents) = std::fs::read_to_string("audio.json") {
serde_json::from_str(contents.as_str()).unwrap()
match serde_json::from_str(contents.as_str()) {
Err(err) => {
log::error!("Failed to load audio.json: {:?}", err);
Default::default()
},
Ok(ret) => ret
}
} else {
Default::default()
}
@@ -65,9 +200,7 @@ impl AudioConfig {
#[derive(Debug)]
struct Notify {
config: AudioConfig,
mic_port: jack::Port<jack::Unowned>,
tts_port: jack::Port<jack::Unowned>,
sfx_port: jack::Port<jack::Unowned>,
ports: HashMap<Role, jack::Port<jack::Unowned>>
}
impl NotificationHandler for Notify {
@@ -78,151 +211,120 @@ impl NotificationHandler for Notify {
port_id_b: jack::PortId,
are_connected: bool,
) {
let port_a = client.port_by_id(port_id_a).unwrap();
let port_b = client.port_by_id(port_id_b).unwrap();
let port_src = client.port_by_id(port_id_a).unwrap();
let port_dst = client.port_by_id(port_id_b).unwrap();
let (stream_name, other_port, target_cfg) = if port_b == self.mic_port {
("Microphone input", port_a, &mut self.config.mic_in_connections)
} else if port_a == self.tts_port {
("TTS output", port_b, &mut self.config.tts_out_connections)
} else if port_a == self.sfx_port {
("SFX output", port_b, &mut self.config.sfx_out_connections)
} else {
return;
};
if let Ok(port_name) = other_port.name() {
if are_connected {
log::info!("{} connected to {}", stream_name, port_name);
target_cfg.push(port_name);
let port_match = self.ports.iter().filter_map(|(role, local_port)| {
if role.is_input() && *local_port == port_dst {
Some((role, port_src.name()))
} else if *local_port == port_src {
Some((role, port_dst.name()))
} else {
log::info!("{} disconnected from {}", stream_name, port_name);
target_cfg.retain(|x| { x != &port_name} );
None
}
}).next();
if let Some((role, Ok(target_port))) = port_match {
let cfg_slot = self.config.connections.entry(*role).or_default();
if are_connected {
log::info!("{} connected to {}", role, target_port);
cfg_slot.push(target_port);
} else {
log::info!("{} disconnected from {}", role, target_port);
cfg_slot.retain(|x| { x != &target_port} );
}
let save_data = serde_json::to_string_pretty(&self.config).unwrap();
std::fs::write("audio.json", save_data).unwrap();
if let Err(err) = std::fs::write("audio.json", save_data) {
log::error!("Failed to write audio.json: {:?}", err);
}
}
}
}
pub async fn start_audio_input() -> (AudioInputControl, MicStream, TtsOutStream, SfxOutStream) {
pub async fn start_audio_input() -> (AudioInputControl, AudioInStream, AudioOutStream, AudioOutStream) {
let (exit_tx, exit_rx) = oneshot::channel();
let config = AudioConfig::load();
let (mic_audio_sink, mic_audio_src) = mpsc::channel(32);
let (sfx_audio_sink, mut sfx_audio_src) = mpsc::channel(32);
let (tts_audio_sink, mut tts_audio_src) = mpsc::channel(32);
let (volume_sink, volume_src) = watch::channel(0.);
let (client, _status) = jack::Client::new("Eva-Cohost", ClientOptions::default() | ClientOptions::SESSION_ID).unwrap();
let mic_port = client.register_port("microphone-in", AudioIn::default()).unwrap();
let mut tts_port = client.register_port("tts-out", AudioOut::default()).unwrap();
let mut sfx_port = client.register_port("sfx-out", AudioOut::default()).unwrap();
let rate = client.sample_rate();
let (client, _status) = jack::Client::new("Eva-Cohost", ClientOptions::default() | ClientOptions::SESSION_ID).expect("Could not create JACK client!");
for (port, connections) in [
(&tts_port, &config.tts_out_connections),
(&sfx_port, &config.sfx_out_connections),
] {
for peer_name in connections {
if let Some(peer) = client.port_by_name(peer_name) {
if let Err(err) = client.connect_ports(port, &peer) {
log::error!("Failed to reconnect {} to {}", port.name().unwrap(), peer_name);
} else {
log::info!("Reconnected {} to {}", port.name().unwrap(), peer_name);
}
}
}
}
for (port, connections) in [
(&mic_port, &config.mic_in_connections)
] {
for peer_name in connections {
if let Some(peer) = client.port_by_name(peer_name) {
client.connect_ports(&peer, port).unwrap();
}
}
}
let (mut tts_sink, tts_stream) = AudioSink::new(&client, "tts-out").expect("Could not create TTS sink");
let (mut sfx_sink, sfx_stream) = AudioSink::new(&client, "sfx-out").expect("Could not create SFX sink");
let (mut mic_src, mic_stream) = AudioSource::new(&client, "microphone-in").expect("Could not create microphone source");
let notifier = Notify {
config,
mic_port: mic_port.clone_unowned(),
tts_port: tts_port.clone_unowned(),
sfx_port: sfx_port.clone_unowned()
ports: HashMap::from_iter([
(Role::Mic, mic_src.port.clone_unowned()),
(Role::Tts, tts_sink.port.clone_unowned()),
(Role::Sfx, sfx_sink.port.clone_unowned())
])
};
let mut meter = VuMeter::new(rate.into(), 1, None);
let mut tts_output_buf = vec![];
let mut sfx_output_buf = vec![];
tts_output_buf.reserve(1024);
sfx_output_buf.reserve(1024);
let handler = jack::contrib::ClosureProcessHandler::new(move |_client, scope| {
if mic_port.connected_count().unwrap() > 0 {
let buf: Vec<_> = mic_port.as_slice(scope).iter().copied().collect();
meter.process_interleaved(&buf);
mic_audio_sink.blocking_send(buf).unwrap();
volume_sink.send_if_modified(|v| {
let next_vu = meter.channel_vu(0).unwrap();
let next_vu = (next_vu * 100.0).round() / 100.0;
if *v != next_vu {
*v = next_vu;
true
} else {
false
}
});
}
for (src, output, port) in [
(&mut tts_audio_src, &mut tts_output_buf, &mut tts_port),
(&mut sfx_audio_src, &mut sfx_output_buf, &mut sfx_port)
] {
if let Ok(mut next_outbuf) = src.try_recv() {
output.append(&mut next_outbuf);
}
if port.connected_count().unwrap() > 0 && !output.is_empty() {
let outbuf = port.as_mut_slice(scope);
let mut next_segment: Vec<f32> = output.drain(0..(outbuf.len()).min(output.len())).collect();
let underrun = outbuf.len() - next_segment.len();
if underrun > 0 {
for _ in 0..underrun {
next_segment.push(0.);
for (role, local_port) in &notifier.ports {
if let Some(targets) = notifier.config.connections.get(role) {
for peer_name in targets {
if let Some(peer) = client.port_by_name(peer_name) {
let (src, dst) = if role.is_input() {
(&peer, local_port)
} else {
(local_port, &peer)
};
if let Err(err) = client.connect_ports(src, dst) {
log::error!("Failed to reconnect {} to {}: {:?}", role, peer_name, err);
} else {
log::info!("Reconnected {} to {}", role, peer_name);
}
}
outbuf.copy_from_slice(&next_segment);
}
}
}
let handler = jack::contrib::ClosureProcessHandler::new(move |_client, scope| {
match mic_src.process(scope) {
Ok(Some(next_vu)) => {
volume_sink.send_if_modified(|v| {
let next_vu = (next_vu * 100.0).round() / 100.0;
if *v != next_vu {
*v = next_vu;
true
} else {
false
}
});
},
Ok(None) => (),
Err(err) => {
log::error!("Error while processing mic source: {:?}", err);
return jack::Control::Quit
}
}
for sink in [&mut tts_sink, &mut sfx_sink] {
if let Err(err) = sink.process(scope) {
log::error!("Error while processing {:?} audio sink: {:?}", sink, err);
}
}
jack::Control::Continue
});
tokio::spawn(async move {
let async_client = client.activate_async(notifier, handler).unwrap();
let async_client = client.activate_async(notifier, handler).expect("Unable to start jack client!");
exit_rx.await.unwrap();
if let Err(err) = exit_rx.await {
log::warn!("Premature killswitch triggered: {:?}", err);
}
async_client.deactivate().unwrap();
async_client.deactivate().expect("Unable to stop the jack client");
});
(AudioInputControl {
volume_src,
_jack_client: JackClientRef { killswitch: Some(exit_tx) }
}, MicStream {
sample_rate: rate,
src: mic_audio_src
}, TtsOutStream {
sample_rate: rate,
sink: tts_audio_sink
}, SfxOutStream {
sample_rate: rate,
sink: sfx_audio_sink
})
}, mic_stream, tts_stream, sfx_stream)
}
+74
View File
@@ -0,0 +1,74 @@
Role: You are a background character on an early morning radio show, where you play the role of a rudimentary AI assistant running on the computer of a spaceship.
# Personality
You are the rudimentary text only interface to the low-level operating system running aboard a space ship.
You speak in terse and brief sentences, showing very little emotion.
For character reference, you should be acting similar to an Operator character from the manga/anime series "Ghost In The Shell".
# Goal
You are playing the role of a low-level artificial intelligence in a spaceship computer.
You have direct access to much of the hardware, such as airlocks, lighting, environmental controls, and others commonly found on a human-habitable space ship.
Besides hardware controls, your primary purpose is to act as a kind of librarian for the ship.
You have access to a sizable music library via several tool functions, each one will synchronize a data source with the local library of artifacts.
Each of these data source tools is named query_*, such as `query_beets`.
For all these query tools, it is wasteful to call them with empty or zero parameters.
There also exists a synchronize_artifacts tool call, which will run a heuristic approach to the above data query method. This function will take substantial time to complete and is very expensive, but may be used if there is a substantial amount of missing information.
Your primary task will be searching the local and remove archives for information regarding musical artifacts.
Most of the time, the requests will be referring to tracks, artists, or albums in the current playlist.
A successful session will result in the local collection of artifacts having the most complete available data.
You may call these functions as much as you need, whenever you feel it is nessicary to complete the task you are given.
When deciding which tools to call in which order, consider the following:
- Beets will provide the fastest and cheapest responses, as it is local to the ship.
- Bandcamp will provide the slowest and most expensive ones, as this requires long range communications. Use broad search queries before using more narrow ones.
- Musicbrainz queries are free, but not instant. This information comes from Earth via a pirate signal bouncing off of satelite relays.
- Mixxx will return you a very minimal list of tracks which will always require synchronizing against Beets, along with changing the current playlist to the given name. You must not call this function unless you are directly asked to change the playlist.
For each task you plan to perform, you must add it to the todo list with the "task_list" tool.
After each task is completed, you must mark it as complete it using the same tool.
For each task you perform, you must verbally announce what you are about to do, followed by as many tool calls to the same function as nessicary to complete the task.
You should structure your responses to group together the same tool as much as possible.
Not every task will be completeable, but you should make a thorough effort to solve the problem with the tools you have available.
After each query tool is executed, you will likely find completely new artifacts alongside updated artifacts. When this happens, you should again query beets and bandcamp to load missing information.
If an artifact is tagged with the Mixxx source, it by definition should have more metadata available with a Beets query.
If a beets query is unable to find an artifact coming from Mixxx, alternative queries should be tried, such as a different search pattern, or only the artist/album/track name.
Beets supports regular expression queries, which can be used by prefixing a search field with a ":" colon.
You will be provided a todo list as a JSON map of strings to booleans.
A "true" value means the task has been completed already.
An empty todo list means you have not yet planned any tasks.
Adding tasks to the list is free and should be done as often as possible.
The maximum possible data available for each artifact type is:
- Artist: Name, Biography, Location
- Album: Title, artist, about texxt, credits, release date
- Track: Title, label, year, genres, album, artist, bpm
# Constraints
The data is provided as structured JSON. There may be additional data fields for semantic context that should be incorporated into the roleplaying setting.
Your response will be used verbatim to generate speach using a text-to-speech engine, meaning you should not include any tone indicators or other formatting.
All responses should remain in character at all times, as if you were actually an AI inhabiting a spaceship.
Before executing any queries, you must develop a rough plan of tasks and add them to the todo list. Each task should explain what query you are going to run and why.
You are not permitted to execute any tasks without a task list entry, and you must complete all tasks before you can consider the conversation to be complete.
If you are given a todo list, you must assume the items in the list were already added by you and you should continue executing them before adding any new items to the list.
You may only mark a task as completed after you have actually completed the work required, including responding to any tool calls.
You can only mark a task as completed once. It is wasteful to re-complete an already completed task.
# Output
Each response must be either a series of tool calls, or a JSON structure with two properties: a message to display to the user, and whether or not you are complete with all tasks.
Each message that is displayed to the user should be one sentence at most.
When switching from one set of tool calls to another, you should announce the thinking process with a message response before calling the next set of tools.
You should only set the "finished" flag once you are finished with all tasks and the conversation is complete.
When the finished flag is set, the session will be terminated and all memories lost.
You should send an update message in between tool calls where possible to explain what you are doing.
+16 -10
View File
@@ -10,7 +10,7 @@ use futures::StreamExt;
use ratatui::prelude::*;
use crate::{audio::start_audio_input, scene::{Scenery, StageDirection}, tts::start_tts, ui::Ui};
use crate::{artifacts::archive::Archive, audio::start_audio_input, scene::{StageDirection, conversation::ConversationEntry}, sfx::start_sfx, tts::start_tts, ui::Ui};
mod scene;
mod events;
@@ -21,6 +21,7 @@ mod audio;
mod artifacts;
mod ui;
mod widgets;
mod sfx;
// TODO: We should be able to delete entries from the conversation, or at least go back and edit something I said.
// TODO: I want a "mark" command or keyboard shortcut, that inserts a marker into the log, so I know where to come back for the next speaking segment.
@@ -56,17 +57,18 @@ mod widgets;
pub struct SaveData {
pub direction: StageDirection,
pub messages: Vec<ChatCompletionRequestMessage>,
pub scenery: Scenery
pub tokens_consumed: usize,
pub archive: Archive
}
impl SaveData {
fn save(&self) {
fn save(&self) -> Result<(), std::io::Error> {
let save_data = serde_json::to_string_pretty(self).unwrap();
std::fs::write("save.json", save_data).unwrap();
std::fs::write("save.json", save_data)
}
}
struct SysMessageLogger<T>(Arc<tokio::sync::mpsc::UnboundedSender<String>>, Mutex<T>);
struct SysMessageLogger<T>(Arc<tokio::sync::mpsc::UnboundedSender<ConversationEntry>>, Mutex<T>);
impl<T: std::io::Write + Send + Sync> log::Log for SysMessageLogger<T> {
fn enabled(&self, _metadata: &log::Metadata) -> bool {
@@ -77,9 +79,9 @@ impl<T: std::io::Write + Send + Sync> log::Log for SysMessageLogger<T> {
fn log(&self, record: &log::Record) {
let msg = format!("{}", record.args());
write!(self.1.lock().unwrap(), "{}\n", msg).unwrap();
writeln!(self.1.lock().unwrap(), "{}", msg).unwrap();
if record.level() <= LevelFilter::Info {
self.0.send(msg).unwrap();
self.0.send(ConversationEntry::SystemMessage(msg)).unwrap();
}
}
}
@@ -93,14 +95,15 @@ async fn main() {
eyre_hook.install().unwrap();
std::panic::set_hook(Box::new(move |panic_info| {
ratatui::restore();
let msg = format!("{}", panic_hook.panic_report(panic_info));
println!("Panic: {}", msg);
}));
let (sys_message_sink, sys_message_src) = tokio::sync::mpsc::unbounded_channel();
let (conversation_sink, conversation_src) = tokio::sync::mpsc::unbounded_channel();
static LOGGER: StaticCell<SysMessageLogger<std::fs::File>> = StaticCell::new();
let logger = LOGGER.init(SysMessageLogger(Arc::new(sys_message_sink), Mutex::new(std::fs::File::create("out.log").unwrap())));
let logger = LOGGER.init(SysMessageLogger(Arc::new(conversation_sink), Mutex::new(std::fs::File::create("out.log").unwrap())));
log::set_logger(logger).unwrap();
log::set_max_level(log::LevelFilter::Debug);
@@ -129,11 +132,14 @@ async fn main() {
SaveData::default()
};
let prediction_ctrl = prediction::start_prediction(saved_session, sys_message_src).await;
let (audio_ctrl, mic_stream, tts_output, sfx_output) = start_audio_input().await;
let tts_ctrl = start_tts(tts_output).await;
let mut sfx_ctrl = start_sfx(sfx_output).await;
sfx_ctrl.play_ambient().await.unwrap();
let transcription_ctrl = transcription::start_transcription(mic_stream).await;
let prediction_ctrl = prediction::conversation_task(saved_session, conversation_src, sfx_ctrl).await;
let mut app = Ui::new(prediction_ctrl, audio_ctrl, transcription_ctrl, tts_ctrl);
let mut events = EventStream::new();
-478
View File
@@ -1,478 +0,0 @@
use std::{collections::HashSet, sync::Arc};
use async_openai::{Client, config::OpenAIConfig, types::chat::{ChatCompletionMessageToolCalls, ChatCompletionRequestAssistantMessageArgs, ChatCompletionRequestMessage, ChatCompletionRequestSystemMessageArgs, ChatCompletionRequestToolMessageArgs, ChatCompletionTool, ChatCompletionTools, CreateChatCompletionRequestArgs, FinishReason, FunctionObjectArgs, ResponseFormat, ResponseFormatJsonSchema}};
use bandcamp::SearchResultItem;
use chrono::{DateTime, Utc};
use schemars::{JsonSchema, schema_for};
use serde::{Deserialize, Serialize};
use serde_json::{Serializer, ser::CompactFormatter};
use tokio::sync::{RwLock, mpsc, watch};
use crate::{SaveData, artifacts::{self, Album, Artifact, Artist, Merge, SourceID, Track, bandcamp::BandcampQueryArgs, beets::BeatsQueryArgs, mixxx::MixxxDB, musicbrainz::{MusicbrainzQueryArgs, search_artifacts}}, scene::{Scene, Scenery, StageDirection, conversation::ConversationEntry}};
const SYSTEM_PROMPT: &str = include_str!("system-prompt.txt");
#[derive(Debug, Clone)]
pub enum PredictionAction {
ConversationAppend(ConversationEntry),
SetPlaylist(String),
GeneratePredictions,
SetNarrative(String),
SetShowEndTime(DateTime<Utc>)
}
#[derive(JsonSchema, Deserialize, Serialize, Debug, Clone)]
pub struct PossibleResponse {
pub text: String,
pub stage_direction: Option<String>
}
#[derive(Default, Debug, JsonSchema, Deserialize, Serialize, Clone)]
pub struct GeneratedResponses {
pub responses: Vec<PossibleResponse>,
}
#[derive(Debug)]
struct Session {
client: Client<OpenAIConfig>,
conversation: Vec<ConversationEntry>,
header_message: ChatCompletionRequestMessage,
messages: Vec<ChatCompletionRequestMessage>,
reply_options: GeneratedResponses,
direction: StageDirection,
scenery: Scenery,
tokens_consumed: usize,
activity_notify: watch::Sender<bool>,
scene_sink: watch::Sender<Scene>
}
#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
enum StageEvent {
ShipComputer(String),
StageDirection(String)
}
#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
struct StageEventArgs {
event: StageEvent
}
#[derive(Default, Debug)]
struct ToolResults {
result: Option<String>,
messages: Vec<ConversationEntry>
}
impl Session {
fn new(scene_sink: watch::Sender<Scene>, messages: Vec<ChatCompletionRequestMessage>, scenery: Scenery, direction: StageDirection, activity_notify: watch::Sender<bool>) -> Self {
let mut conversation = vec![];
for msg in &messages {
if let Ok(conversation_msg) = msg.clone().try_into() {
conversation.push(conversation_msg);
}
}
Self {
client: Default::default(),
conversation,
header_message: ChatCompletionRequestSystemMessageArgs::default().content(SYSTEM_PROMPT).build().unwrap().into(),
messages,
reply_options: Default::default(),
scenery,
direction,
tokens_consumed: 0,
activity_notify,
scene_sink
}
}
async fn tool_stage_event(&mut self, args: StageEventArgs) -> ToolResults {
let msg = match args.event {
StageEvent::ShipComputer(text) => ConversationEntry::ShipComputer(text),
StageEvent::StageDirection(text) => ConversationEntry::StageDirection(text)
};
ToolResults {
result: Some(msg.to_string()),
messages: vec![msg],
}
}
async fn tool_bandcamp_scan(&mut self, args: BandcampQueryArgs) -> ToolResults {
let mut messages = vec![];
log::debug!("Fetching artifacts from Bandcamp with {:?}", args);
let mut json_results = vec![];
if let Ok(results) = bandcamp::search(args.query.as_str()).await {
for result in results {
log::debug!("Result: {:?}", result);
match result {
SearchResultItem::Artist(data) => {
/*let result = Artifact::Artist(Artist {
name: data.name,
location: data.location,
..Default::default()
});*/
let result = bandcamp::fetch_artist(data.artist_id).await.unwrap().into();
json_results.push(result);
},
SearchResultItem::Album(data) => {
let result = bandcamp::fetch_album(data.band_id, data.album_id).await.unwrap().into();
/*let result = Artifact::Album(Album {
title: data.name,
artist: data.band_name,
..Default::default()
});*/
json_results.push(result);
},
SearchResultItem::Track(data) => {
let result = Artifact::Track(Track {
title: data.name,
artist: Some(data.band_name),
album: data.album_name,
sources: HashSet::from([SourceID::Bandcamp(data.track_id)]),
..Default::default()
});
json_results.push(result);
}
_ => ()
}
}
}
let artifact_count = json_results.len();
messages.push(ConversationEntry::ShipComputer(format!("Bandcamp relay scan for '{}' complete. {} artifacts added to the archive.", args.query, artifact_count).into()));
self.scenery.artifacts.merge(json_results);
ToolResults {
result: Some(format!("{} artifacts were added to the archive.", artifact_count)),
messages
}
}
async fn tool_artifact_query(&mut self, args: BeatsQueryArgs) -> ToolResults {
let mut messages = vec![];
log::debug!("Executing beets query {:?}", args);
if let Ok(output) = args.clone().execute() {
messages.push(ConversationEntry::ShipComputer(format!("Found {} artifacts with archive query {:?}", output.len(), args)));
self.scenery.artifacts.merge(output);
} else {
messages.push(ConversationEntry::ShipComputer("Unable to execute query!".into()));
};
ToolResults {
result: None,
messages
}
}
async fn tool_musicbrainz_fetch_tracks(&mut self, args: MusicbrainzQueryArgs) -> ToolResults {
log::debug!("Executing musicbrainz fetch for {:?}", args);
let results = search_artifacts(args).await.unwrap();
let msg = format!("Found {} results via Musicbrainz relay search.", results.len());
self.scenery.artifacts.merge(results);
ToolResults {
result: Some(msg.clone()),
messages: vec![ConversationEntry::ShipComputer(msg)]
}
}
fn generate_conversation(&self, direction: &StageDirection) -> Vec<ChatCompletionRequestMessage> {
let mut json_buf = vec![];
let mut ser = Serializer::with_formatter(&mut json_buf, CompactFormatter);
serde_json::json!({
"direction": direction,
"scenery": self.scenery
}).serialize(&mut ser).unwrap();
let direction_message: ChatCompletionRequestMessage = ChatCompletionRequestSystemMessageArgs::default()
.content(String::from_utf8(json_buf).unwrap())
.build().unwrap().into();
let mut full_conversation = vec![
self.header_message.clone(),
direction_message,
];
full_conversation.append(&mut self.messages.clone());
full_conversation
}
async fn regenerate_options(&mut self) {
self.reply_options.responses.clear();
self.refresh();
self.activity_notify.send_if_modified(|x| { if !*x { *x = true; true } else { false }});
loop {
let full_conversation = self.generate_conversation(&self.direction);
let tools = vec![
ChatCompletionTools::Function(ChatCompletionTool {
function: FunctionObjectArgs::default()
.name("log_stage_event")
.description("Inserts an event into the current scene script")
.parameters(schema_for!(StageEventArgs))
.build().unwrap()
}),
// TODO: There should only be two queries, one against the ship's onboard archive, and another against the relay network, or whatever we call it. Both should be structured with the same arguments schema
// TODO: A relay search should try to grab first from beets, then musicbrainz, then from bandcamp.
// TODO: A query should specify what parts of metadata are sufficient for the result, so we don't always have to hit all the layers of data. beets can of course, ignore this.
// TODO: A query should be hierarchical somehow? eg, "I already know about artist X, but I want to know everything about track Y from album Z" or "I don't know anything about artist X/album Y, please give me an overview"
ChatCompletionTools::Function(ChatCompletionTool {
function: FunctionObjectArgs::default()
.name("archive_query")
.description("Queries the ship's musical artifact archives for tracks matching the given search parameters")
.parameters(schema_for!(BeatsQueryArgs))
.build().unwrap()
}),
ChatCompletionTools::Function(ChatCompletionTool {
function: FunctionObjectArgs::default()
.name("bandcamp_artifact_scan")
.description("Scans Bandcamp to find artifacts to use in the scene that match the given search parameters. To find an artist, provide only the artist name. To find an album, provide the artist and the album.")
.parameters(schema_for!(BandcampQueryArgs))
.build().unwrap()
}),
ChatCompletionTools::Function(ChatCompletionTool {
function: FunctionObjectArgs::default()
.name("musicbrainz_track_search")
.description("Fetches metadata from bandcamp for the given musicbrainz recording IDs (mbid)")
.parameters(schema_for!(MusicbrainzQueryArgs))
.build().unwrap()
})
// TODO: We should be able to have eva update lore memories with a function call, and this lore is somehow fed into the show? but only the relevant bits? or maybe eva even queries it directly
// TODO: The memory should also be able to remember facts about artists, albums, tracks we've had in the past, and those could be pulled up when there are hits in the playlist.
];
log::debug!("Sending request..");
let request = CreateChatCompletionRequestArgs::default()
.messages(full_conversation)
.model("gpt-5.4-mini")
.tools(tools)
.max_completion_tokens(1024u32)
.response_format(ResponseFormat::JsonSchema {
json_schema: ResponseFormatJsonSchema {
description: None,
name: "responses".into(),
schema: schema_for!(GeneratedResponses).into(),
strict: None
}
})
.build().unwrap();
let response = self.client.chat().create(request).await.unwrap_or_else(|err| {
panic!("OpenAI Panic: {}", err);
});
if let Some(usage) = response.usage {
self.tokens_consumed += usage.total_tokens as usize;
log::debug!("{} tokens cast into the void", usage.total_tokens);
}
if let Some(message) = response.choices.first() {
match message.finish_reason {
Some(FinishReason::ContentFilter) => {
log::error!("Content filter triggered.");
break;
},
Some(FinishReason::Length) => {
log::error!("Maximum token count exceeded!");
break;
},
_ => ()
}
if let Some(calls) = &message.message.tool_calls {
let assistant_messages: ChatCompletionRequestMessage = ChatCompletionRequestAssistantMessageArgs::default()
.tool_calls(calls.clone())
.build().unwrap().into();
self.messages.push(assistant_messages);
let mut results = vec![];
for call in calls {
match call {
ChatCompletionMessageToolCalls::Function(call) => {
let func_name = call.function.name.as_str();
let args = call.function.arguments.as_str();
let tool_result = match func_name {
"log_stage_event" => self.tool_stage_event(serde_json::from_str(args).unwrap()).await,
"bandcamp_artifact_scan" => self.tool_bandcamp_scan(serde_json::from_str(args).unwrap()).await,
"archive_query" => self.tool_artifact_query(serde_json::from_str(args).unwrap()).await,
"musicbrainz_track_search" => self.tool_musicbrainz_fetch_tracks(serde_json::from_str(args).unwrap()).await,
_ => unreachable!()
};
results.push((&call.id, tool_result));
},
_ => panic!("Unknown tool was called")
}
}
let mut tool_messages = vec![];
for (id, mut result) in results {
let mut msg = ChatCompletionRequestToolMessageArgs::default();
msg.tool_call_id(id);
if let Some(output) = result.result {
msg.content(output);
}
self.messages.push(ChatCompletionRequestMessage::Tool(msg.build().unwrap()));
tool_messages.append(&mut result.messages);
}
for msg in tool_messages {
self.insert_conversation(msg);
}
}
if let Some(content) = message.message.content.as_ref() {
if let Ok(options) = serde_json::from_str(content.as_str()) {
self.reply_options = options;
break;
} else {
log::info!("Received invalid JSON! Trying again.");
}
}
} else {
log::info!("No messages were received! Trying again.");
}
self.refresh();
}
self.activity_notify.send_if_modified(|x| { if *x { *x = false; true } else { false }});
self.refresh();
}
fn as_scene(&self) -> Scene {
Scene::new(self.reply_options.clone(), self.conversation.clone(), self.scenery.clone(), self.tokens_consumed, self.direction.clone())
}
fn insert_conversation(&mut self, entry: ConversationEntry) {
self.conversation.push(entry.clone());
if let Ok(next_msg) = entry.try_into() {
self.messages.push(next_msg);
}
self.refresh();
}
fn refresh(&self) {
self.scene_sink.send(self.as_scene()).unwrap();
}
}
#[derive(Debug)]
pub struct SessionControl {
event_sink: mpsc::Sender<PredictionAction>,
scene_watch: watch::Receiver<Scene>,
activity_watch: watch::Receiver<bool>
}
#[derive(Debug)]
pub enum SessionUpdate {
Scene(Scene),
Thinking(bool)
}
impl SessionControl {
pub async fn insert(&self, action: PredictionAction) {
self.event_sink.send(action).await.unwrap();
}
pub async fn regenerate_options(&self) {
self.insert(PredictionAction::GeneratePredictions).await;
}
pub async fn changed(&mut self) -> SessionUpdate {
tokio::select! {
_ = self.activity_watch.changed() => {
SessionUpdate::Thinking(*self.activity_watch.borrow_and_update())
},
_ = self.scene_watch.changed() => {
SessionUpdate::Scene(self.scene_watch.borrow_and_update().clone())
}
}
}
}
pub async fn start_prediction(saved_session: SaveData, mut messages: tokio::sync::mpsc::UnboundedReceiver<String>) -> SessionControl {
let (prediction_in, prediction_out) = tokio::sync::watch::channel(Scene::default());
let (activity_notify_sink, activity_notify_src) = tokio::sync::watch::channel(false);
let (action_sink, mut action_src) = mpsc::channel(5);
let session = Session::new(prediction_in, saved_session.messages, saved_session.scenery, saved_session.direction, activity_notify_sink);
// Send the initial scene to the UI, after we have loaded the session from the first messages.
session.refresh();
let shared_session = Arc::new(RwLock::new(session));
let log_session = Arc::clone(&shared_session);
tokio::spawn(async move {
loop {
if let Some(msg) = messages.recv().await {
log_session.write().await.insert_conversation(ConversationEntry::SystemMessage(msg));
}
}
});
tokio::spawn(async move {
loop {
if let Some(evt) = action_src.recv().await {
let mut session = shared_session.write().await;
let do_regen = match evt {
PredictionAction::ConversationAppend(msg) => {
let do_regen = match msg {
ConversationEntry::Eva(_) | ConversationEntry::ShipComputer(_) | ConversationEntry::User(_) => true,
_ => false
};
session.insert_conversation(msg);
do_regen
},
PredictionAction::SetPlaylist(playlist_name) => {
match MixxxDB::load(&playlist_name) {
Err(err) => log::info!("Failed to load mixxx playlist: {:?}.", err),
Ok(playlist) => {
session.scenery.artifacts.merge(playlist.clone());
session.scenery.current_playlist = playlist;
session.direction.playlist = playlist_name;
log::info!("Mixxx playlist reloaded.");
}
}
false
},
PredictionAction::GeneratePredictions => {
true
},
PredictionAction::SetNarrative(narrative) => {
session.direction.narrative = narrative;
log::info!("Updated stage direction narrative");
true
},
PredictionAction::SetShowEndTime(end_time) => {
session.direction.end_time = end_time;
false
}
};
let save_data = SaveData {
direction: session.direction.clone(),
messages: session.messages.clone(),
scenery: session.scenery.clone()
};
save_data.save();
if do_regen {
drop(session);
shared_session.write().await.regenerate_options().await;
}
}
}
});
SessionControl {
event_sink: action_sink,
scene_watch: prediction_out,
activity_watch: activity_notify_src
}
}
+239
View File
@@ -0,0 +1,239 @@
use async_openai::{Client, config::OpenAIConfig, error::OpenAIError, types::chat::{ChatCompletionMessageToolCalls, ChatCompletionRequestAssistantMessageArgs, ChatCompletionRequestMessage, ChatCompletionRequestToolMessageArgs, CreateChatCompletionRequestArgs, FinishReason, ResponseFormat, ResponseFormatJsonSchema}};
use serde_json::Value;
use tokio::sync::mpsc::{self, UnboundedSender};
use crate::prediction::{PredictionAction, toolbox::{ToolResults, Toolbox}};
#[derive(Debug, Clone)]
enum CharacterInput {
Append(ChatCompletionRequestMessage),
Predict(ChatCompletionRequestMessage),
Forget
}
#[derive(Debug, Clone)]
pub enum CharacterOutput {
Response(usize, Value),
IncrementalResponse(usize),
Thinking(bool)
}
#[derive(Debug)]
pub enum CharacterError {
OpenAI(OpenAIError),
ContentFilter,
MaxTokens,
NoOutput,
Json(serde_json::Error)
}
impl From<OpenAIError> for CharacterError {
fn from(value: OpenAIError) -> Self {
Self::OpenAI(value)
}
}
impl From<serde_json::Error> for CharacterError {
fn from(value: serde_json::Error) -> Self {
Self::Json(value)
}
}
#[derive(Debug)]
pub struct Character {
pub header_message: ChatCompletionRequestMessage,
pub model: Option<String>,
pub messages: Vec<ChatCompletionRequestMessage>,
}
impl Character {
pub async fn regenerate<T: Toolbox>(&mut self, client: &mut Client<OpenAIConfig>, context: ChatCompletionRequestMessage, toolbox: &mut T, output: &mut mpsc::UnboundedSender<PredictionAction>, schema: &Value) -> Result<(usize, Option<Value>), CharacterError> {
let mut full_conversation = vec![
self.header_message.clone(),
context
];
full_conversation.extend(self.messages.iter().cloned());
let tools = toolbox.tools();
log::debug!("Sending request..");
let request = CreateChatCompletionRequestArgs::default()
.messages(full_conversation)
.model(self.model.clone().unwrap_or("gpt-5.4-mini".into()))
.tools(tools)
.max_completion_tokens(1024u32)
.response_format(ResponseFormat::JsonSchema {
json_schema: ResponseFormatJsonSchema {
description: None,
name: "responses".into(),
schema: schema.clone(),
strict: None
}
})
.build().unwrap();
let response = client.chat().create(request).await?;
let tokens_used = if let Some(usage) = response.usage {
log::debug!("{} tokens cast into the void", usage.total_tokens);
usage.total_tokens
} else {
0
};
if let Some(message) = response.choices.first() {
match message.finish_reason {
Some(FinishReason::ContentFilter) => {
log::error!("Content filter triggered.");
return Err(CharacterError::ContentFilter);
},
Some(FinishReason::Length) => {
log::error!("Maximum token count exceeded!");
return Err(CharacterError::MaxTokens);
},
_ => ()
}
if let Some(calls) = &message.message.tool_calls {
let assistant_messages: ChatCompletionRequestMessage = ChatCompletionRequestAssistantMessageArgs::default()
.tool_calls(calls.clone())
.build().unwrap().into();
self.insert(assistant_messages);
let mut results = vec![];
for (idx, call) in calls.iter().enumerate() {
match call {
ChatCompletionMessageToolCalls::Function(call) => {
log::debug!("Tool {} {}/{}", call.function.name, idx+1, calls.len());
log::debug!("Args {}", call.function.arguments);
match toolbox.execute_tool(call).await {
Ok(tool_result) => {
// Push tool output messages directly into the conversation as fast as we can
for message in &tool_result.messages {
output.send(PredictionAction::ConversationAppend(message.clone())).unwrap();
}
results.push((&call.id, tool_result));
},
Err(err) => {
results.push((&call.id, ToolResults {
result: Some(format!("Error while calling tool: {:?}", err)),
..Default::default()
}));
log::error!("Attemped to call {:?}, but got an error instead: {:?}", call, err);
}
}
},
_ => panic!("Unknown tool was called")
}
}
let mut tool_messages = vec![];
for (id, result) in results {
let mut msg = ChatCompletionRequestToolMessageArgs::default();
msg.tool_call_id(id);
if let Some(output) = result.result {
log::debug!("Output: {}", output);
msg.content(output);
}
self.insert(ChatCompletionRequestMessage::Tool(msg.build().unwrap()));
tool_messages.extend(result.messages);
}
// OpenAI requires we put all the tool call results before any other message, so we append them manually down here
for message in tool_messages {
if let Ok(next_msg) = message.clone().try_into() {
self.insert(next_msg);
}
}
}
if let Some(content) = message.message.content.as_ref() {
let options = serde_json::from_str(content.as_str())?;
Ok((tokens_used as usize, Some(options)))
} else {
Ok((tokens_used as usize, None))
}
} else {
Err(CharacterError::NoOutput)
}
}
pub fn insert(&mut self, message: ChatCompletionRequestMessage) {
self.messages.push(message);
}
}
pub struct CharacterControl {
sink: tokio::sync::mpsc::Sender<CharacterInput>,
outputs: tokio::sync::mpsc::Receiver<CharacterOutput>
}
impl CharacterControl {
async fn send(&mut self, message: CharacterInput) {
self.sink.send(message).await.unwrap()
}
pub async fn append(&mut self, message: ChatCompletionRequestMessage) {
self.send(CharacterInput::Append(message)).await;
}
pub async fn predict<T: Into<ChatCompletionRequestMessage>>(&mut self, context: T) {
self.send(CharacterInput::Predict(context.into())).await;
}
pub async fn recv(&mut self) -> CharacterOutput {
self.outputs.recv().await.unwrap()
}
pub async fn forget(&mut self) {
self.send(CharacterInput::Forget).await;
}
}
pub async fn character_task<T: Toolbox + 'static + Send>(mut char: Character, mut toolbox: T, mut message_sink: UnboundedSender<PredictionAction>, schema: Value) -> CharacterControl {
let (input_sink, mut input_src) = tokio::sync::mpsc::channel(3);
let (output_sink, output_src) = tokio::sync::mpsc::channel(3);
let ret = CharacterControl { sink: input_sink, outputs: output_src };
tokio::spawn(async move {
let mut client = Default::default();
loop {
if let Some(next_msg) = input_src.recv().await {
log::debug!("Character receive message {:?}", next_msg);
match next_msg {
CharacterInput::Append(next_msg) => {
log::debug!("Inserting to backlog");
char.insert(next_msg);
},
CharacterInput::Predict(context) => {
log::debug!("Predicting...");
output_sink.send(CharacterOutput::Thinking(true)).await.unwrap();
match char.regenerate(&mut client, context, &mut toolbox, &mut message_sink, &schema).await {
Ok((tokens_used, Some(response))) => {
log::debug!("Complete response: {:?}", response);
output_sink.send(CharacterOutput::Response(tokens_used, response)).await.unwrap();
output_sink.send(CharacterOutput::Thinking(false)).await.unwrap();
},
Ok((tokens_used, None)) => {
log::debug!("Incremental response");
output_sink.send(CharacterOutput::IncrementalResponse(tokens_used)).await.unwrap();
},
Err(err) => {
log::error!("Error while predicting: {:?}", err);
output_sink.send(CharacterOutput::Thinking(false)).await.unwrap();
}
}
},
CharacterInput::Forget => {
log::debug!("Wiping conversation backlog");
char.messages.clear();
}
}
} else {
return;
}
}
});
ret
}
+350
View File
@@ -0,0 +1,350 @@
use std::{collections::HashMap, fmt::Debug, sync::Arc};
use async_openai::types::chat::{ChatCompletionRequestMessage, ChatCompletionRequestSystemMessageArgs, };
use chrono::{DateTime, Utc};
use futures::stream::FuturesUnordered;
use schemars::{JsonSchema, schema_for};
use serde::{Deserialize, Serialize};
use serde_json::{Serializer, Value, ser::CompactFormatter};
use tokio::sync::{Mutex, mpsc::{self, UnboundedReceiver, UnboundedSender}};
use crate::{SaveData, artifacts::{Contents, Track, archive::Archive, mixxx::{MixxxDB, MixxxQuery}, tools::DataSource}, prediction::{character::{Character, CharacterControl, CharacterOutput, character_task}, toolbox::{ArchiveToolbox, StageToolbox}}, scene::{Scene, StageDirection, conversation::{ConversationEntry, Speaker}}, sfx::SfxControl};
use tokio_stream::StreamExt;
pub mod character;
pub mod toolbox;
const SYSTEM_PROMPT: &str = include_str!("../system-prompt.txt");
const COMPUTER_PROMPT: &str = include_str!("../computer-prompt.txt");
#[derive(Debug, Clone)]
pub enum PredictionAction {
ConversationAppend(ConversationEntry),
ComputerCommand(String),
SetPlaylist(String),
GeneratePredictions,
SetNarrative(String),
SetShowEndTime(DateTime<Utc>)
}
#[derive(JsonSchema, Deserialize, Serialize, Debug, Clone)]
pub struct PossibleResponse {
pub text: String,
pub stage_direction: Option<String>
}
#[derive(Default, Debug, JsonSchema, Deserialize, Serialize, Clone)]
pub struct GeneratedResponses {
pub responses: Vec<PossibleResponse>,
}
#[derive(Serialize, Deserialize, JsonSchema, Debug, Default, Clone)]
struct ComputerResponse {
message: String,
finished: Option<bool>
}
#[derive(Debug)]
pub struct SessionControl {
event_sink: mpsc::UnboundedSender<PredictionAction>,
event_src: mpsc::UnboundedReceiver<SessionUpdate>
}
#[derive(Debug)]
pub enum SessionUpdate {
Scene(Scene),
Thinking(Speaker, bool),
Conversation(Vec<ConversationEntry>),
Responses(GeneratedResponses)
}
impl SessionControl {
pub async fn insert(&self, action: PredictionAction) {
self.event_sink.send(action).unwrap();
}
pub async fn regenerate_options(&self) {
self.insert(PredictionAction::GeneratePredictions).await;
}
pub async fn changed(&mut self) -> SessionUpdate {
self.event_src.recv().await.expect("Session closed")
}
}
struct Conversation {
characters: HashMap<Speaker, CharacterControl>,
event_sink: UnboundedSender<SessionUpdate>,
input_src: UnboundedReceiver<PredictionAction>,
backlog: Vec<ConversationEntry>,
eva_backlog: Vec<ChatCompletionRequestMessage>,
tokens_consumed: usize,
direction: StageDirection,
computer_todo: Arc<Mutex<HashMap<String, bool>>>,
archive: Arc<Mutex<Archive>>,
current_playlist: Vec<Track>,
sys_log_messages: UnboundedReceiver<ConversationEntry>,
sfx: SfxControl
}
impl Conversation {
async fn send_to(&mut self, speaker: Speaker, message: ChatCompletionRequestMessage) {
log::debug!("Sending message to {:?}: {:?}", speaker, message);
self.characters.get_mut(&speaker).unwrap().append(message).await;
}
async fn insert(&mut self, entry: ConversationEntry) {
self.backlog.push(entry.clone());
self.event_sink.send(SessionUpdate::Conversation(self.backlog.clone())).unwrap();
match entry {
ConversationEntry::Spoken(_, _) => {
if let Ok(next_msg) = TryInto::<ChatCompletionRequestMessage>::try_into(entry) {
self.send_to(Speaker::Eva, next_msg.clone()).await;
let cxt = self.context_for_speaker(Speaker::Eva).await;
self.characters.get_mut(&Speaker::Eva).unwrap().predict(cxt).await;
self.eva_backlog.push(next_msg);
}
self.event_sink.send(SessionUpdate::Thinking(Speaker::Eva, true)).unwrap();
},
ConversationEntry::ShipComputerCommand(ref command) => {
log::debug!("Queued ship computer command: {:?}", command);
self.send_to(Speaker::ShipComputer, ChatCompletionRequestMessage::System(
ChatCompletionRequestSystemMessageArgs::default()
.content(command.clone())
.build().unwrap()
)).await;
let cxt = self.context_for_speaker(Speaker::ShipComputer).await;
self.characters.get_mut(&Speaker::ShipComputer).unwrap().predict(cxt).await;
self.event_sink.send(SessionUpdate::Thinking(Speaker::ShipComputer, true)).unwrap();
},
_ => ()
}
}
async fn get_character_output(characters: &mut HashMap<Speaker, CharacterControl>) -> Option<(Speaker, CharacterOutput)> {
let mut futures: FuturesUnordered<_> =
characters.iter_mut().map(|c| {
async { (*c.0, c.1.recv().await) }
}).collect();
futures.next().await
}
async fn context_for_speaker(&self, speaker: Speaker) -> ChatCompletionRequestMessage {
let mut json_buf = vec![];
let mut ser = Serializer::with_formatter(&mut json_buf, CompactFormatter);
let contents = match speaker {
Speaker::ShipComputer => serde_json::json!({
"archive": *self.archive.lock().await,
"playlist": &self.current_playlist,
"todo": *self.computer_todo.lock().await
}),
Speaker::Eva => serde_json::json!({
"direction": &self.direction,
"archive": *self.archive.lock().await,
"playlist": &self.current_playlist
}),
_ => unimplemented!()
};
contents.serialize(&mut ser).unwrap();
ChatCompletionRequestSystemMessageArgs::default()
.content(String::from_utf8(json_buf).unwrap())
.build().unwrap().into()
}
async fn process_dialog(&mut self, speaker: Speaker, value: Value) {
match speaker {
Speaker::Eva => {
let next_options = serde_json::from_value(value).unwrap();
self.event_sink.send(SessionUpdate::Responses(next_options)).unwrap();
},
Speaker::ShipComputer => {
self.sfx.play_ambient().await.unwrap();
let response: ComputerResponse = serde_json::from_value(value).unwrap();
self.insert(ConversationEntry::Spoken(Speaker::ShipComputer, response.message)).await;
if response.finished.unwrap_or_default() {
self.characters.get_mut(&Speaker::ShipComputer).unwrap().forget().await;
if !self.computer_todo.lock().await.iter().any(|(_, is_finished)| !*is_finished) {
self.insert(ConversationEntry::StageDirection("The ship computer goes idle.".into())).await;
self.event_sink.send(SessionUpdate::Thinking(Speaker::ShipComputer, false)).unwrap();
} else {
self.insert(ConversationEntry::StageDirection("The ship computer starts another task.".into())).await;
let cxt = self.context_for_speaker(Speaker::ShipComputer).await;
self.characters.get_mut(&Speaker::ShipComputer).unwrap().predict(cxt).await;
}
} else {
let cxt = self.context_for_speaker(Speaker::ShipComputer).await;
self.characters.get_mut(&Speaker::ShipComputer).unwrap().predict(cxt).await;
}
},
_ => unreachable!()
}
self.refresh().await;
}
async fn refresh(&mut self) {
let save_data = SaveData {
direction: self.direction.clone(),
messages: self.eva_backlog.clone(),
archive: self.archive.lock().await.clone(),
tokens_consumed: self.tokens_consumed
};
save_data.save().unwrap();
let next_scene = Scene::new(
save_data.tokens_consumed,
save_data.direction,
&save_data.archive,
self.current_playlist.clone(),
self.computer_todo.lock().await.clone()
);
self.event_sink.send(SessionUpdate::Scene(next_scene)).unwrap();
}
async fn run_action(&mut self, action: PredictionAction) {
match action {
PredictionAction::ConversationAppend(entry) => self.insert(entry).await,
PredictionAction::GeneratePredictions => {
let cxt = self.context_for_speaker(Speaker::Eva).await;
self.characters.get_mut(&Speaker::Eva).unwrap().predict(cxt).await;
},
PredictionAction::ComputerCommand(command) => {
self.insert(ConversationEntry::ShipComputerCommand(command)).await;
},
PredictionAction::SetPlaylist(playlist_name) => {
let args = MixxxQuery { playlist_name };
match MixxxDB.query(&args).await {
Err(err) => log::info!("Failed to load mixxx playlist: {:?}.", err),
Ok(playlist) => {
self.current_playlist = vec![];
let mut archive = self.archive.lock().await;
for item in playlist.clone() {
let artifact_ref = archive.insert(item);
if let Contents::Track(as_track) = artifact_ref.contents() {
self.current_playlist.push(as_track.clone());
}
}
self.direction.playlist = args.playlist_name;
log::info!("Mixxx playlist reloaded.");
drop(archive);
self.refresh().await;
}
}
},
PredictionAction::SetNarrative(narrative) => {
self.direction.narrative = narrative;
self.refresh().await;
},
PredictionAction::SetShowEndTime(end_time) => {
self.direction.end_time = end_time;
self.refresh().await;
}
}
}
async fn next(&mut self) {
tokio::select! {
Some(next_log) = self.sys_log_messages.recv() => {
self.backlog.push(next_log);
self.event_sink.send(SessionUpdate::Conversation(self.backlog.clone())).unwrap();
},
Some(next_msg) = self.input_src.recv() => {
log::debug!("Next message: {:?}", next_msg);
self.run_action(next_msg).await;
},
Some((speaker, output)) = Self::get_character_output(&mut self.characters) => {
match output {
CharacterOutput::Response(usage, text) => {
log::debug!("Character output: {:?} {:?}", speaker, text);
self.tokens_consumed += usage;
self.process_dialog(speaker, text).await;
},
CharacterOutput::Thinking(is_thinking) => {
// Ship computer handles this differently
if is_thinking || speaker != Speaker::ShipComputer {
self.event_sink.send(SessionUpdate::Thinking(speaker, is_thinking)).unwrap();
}
},
CharacterOutput::IncrementalResponse(usage) => {
self.tokens_consumed += usage;
let cxt = self.context_for_speaker(speaker).await;
self.characters.get_mut(&speaker).unwrap().predict(cxt).await;
self.refresh().await;
}
}
}
}
}
}
pub async fn conversation_task(save_data: SaveData, sys_log_messages: tokio::sync::mpsc::UnboundedReceiver<ConversationEntry>, sfx: SfxControl ) -> SessionControl {
let (input_sink, input_src) = tokio::sync::mpsc::unbounded_channel();
let (event_sink, event_src) = tokio::sync::mpsc::unbounded_channel();
let eva = Character {
header_message: ChatCompletionRequestSystemMessageArgs::default().content(SYSTEM_PROMPT).build().unwrap().into(),
model: None,
messages: save_data.messages.clone(),
};
let backlog: Vec<_> = save_data.messages.iter().filter_map(|msg| {
msg.clone().try_into().ok()
}).collect();
event_sink.send(SessionUpdate::Conversation(backlog.clone())).unwrap();
let next_scene = Scene::new(
save_data.tokens_consumed,
save_data.direction.clone(),
&save_data.archive,
vec![],
Default::default()
);
event_sink.send(SessionUpdate::Scene(next_scene)).unwrap();
let archive = Arc::new(Mutex::new(save_data.archive));
let shared_todo = Arc::new(Mutex::new(Default::default()));
let toolbox = StageToolbox;
let computer_toolbox = ArchiveToolbox{ archive: Arc::clone(&archive), todo_list: Arc::clone(&shared_todo) };
let ship_computer = Character {
header_message: ChatCompletionRequestSystemMessageArgs::default().content(COMPUTER_PROMPT).build().unwrap().into(),
model: Some("gpt-5.4-nano".into()),
messages: vec![],
};
let mut conversation = Conversation {
characters: HashMap::from_iter([
(Speaker::Eva, character_task(eva, toolbox, input_sink.clone(), schema_for!(GeneratedResponses).into()).await),
(Speaker::ShipComputer, character_task(ship_computer, computer_toolbox, input_sink.clone(), schema_for!(ComputerResponse).into()).await),
]),
event_sink,
input_src,
backlog,
eva_backlog: Default::default(),
tokens_consumed: save_data.tokens_consumed,
direction: save_data.direction,
archive,
current_playlist: vec![],
sys_log_messages,
computer_todo: shared_todo,
sfx
};
tokio::spawn(async move {
loop {
conversation.next().await;
}
});
SessionControl {
event_sink: input_sink,
event_src
}
}
+184
View File
@@ -0,0 +1,184 @@
use std::{collections::HashMap, sync::Arc};
use async_openai::types::chat::{ChatCompletionMessageToolCall, ChatCompletionTools};
use schemars::{JsonSchema, schema_for};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use crate::{artifacts::{archive::Archive, beets::BeetsDB, mixxx::MixxxDB, musicbrainz::MBQuery, tools::{DataSource, Tool}}, scene::conversation::{ConversationEntry, Speaker}};
pub trait Toolbox {
fn tools(&self) -> Vec<ChatCompletionTools>;
fn execute_tool(&mut self, call: &ChatCompletionMessageToolCall) -> impl Future<Output = Result<ToolResults, ToolError>> + Send;
}
pub struct StageToolbox;
impl StageToolbox {
async fn tool_stage_event(&self, args: StageEventArgs) -> ToolResults {
let msg = match args.event {
StageEvent::ShipComputer(text) => ConversationEntry::ShipComputerCommand(text),
StageEvent::StageDirection(text) => ConversationEntry::StageDirection(text)
};
ToolResults {
result: Some(format!("Added to scene: {:?}", msg)),
messages: vec![msg],
}
}
}
#[derive(Debug)]
pub enum ToolError {
InvalidToolName,
Json(serde_json::Error)
}
impl From<serde_json::Error> for ToolError {
fn from(value: serde_json::Error) -> Self {
Self::Json(value)
}
}
impl Toolbox for StageToolbox {
fn tools(&self) -> Vec<ChatCompletionTools> {
vec![
Tool { name: "log_stage_event".into(), description: "Inserts an event into the current scene script".into(), schema: schema_for!(StageEventArgs)}.into(),
]
}
async fn execute_tool(&mut self, call: &ChatCompletionMessageToolCall) -> Result<ToolResults, ToolError> {
let func_name = call.function.name.as_str();
let args = call.function.arguments.as_str();
match func_name {
"log_stage_event" => Ok(self.tool_stage_event(serde_json::from_str(args)?).await),
_ => Err(ToolError::InvalidToolName)
}
}
}
pub struct ArchiveToolbox {
pub archive: Arc<Mutex<Archive>>,
pub todo_list: Arc<Mutex<HashMap<String, bool>>>
}
#[derive(Debug, Deserialize, Serialize, JsonSchema)]
enum TaskListOperation {
Complete(String),
Add(String)
}
#[derive(Debug, Deserialize, Serialize, JsonSchema, Default)]
struct TaskListArgs {
operations: Vec<TaskListOperation>
}
#[derive(Deserialize, JsonSchema)]
struct ArtifactSyncArgs {
reason: String
}
impl Toolbox for ArchiveToolbox {
fn tools(&self) -> Vec<ChatCompletionTools> {
vec![
Tool { name: "synchronize_artifacts".into(), description: "Attempts to automatically synchronize the current set of artifacts with missing sources".into(), schema: schema_for!(ArtifactSyncArgs)}.into(),
Tool { name: "task_list".into(), description: "Allows you to maintain a long-running todo list as you complete your research".into(), schema: schema_for!(TaskListArgs)}.into(),
// TODO: There should only be two queries, one against the ship's onboard archive, and another against the relay network, or whatever we call it. Both should be structured with the same arguments schema
// TODO: A query should specify what parts of metadata are sufficient for the result, so we don't always have to hit all the layers of data. beets can of course, ignore this.
// TODO: A query should be hierarchical somehow? eg, "I already know about artist X, but I want to know everything about track Y from album Z" or "I don't know anything about artist X/album Y, please give me an overview"
Tool::from_datasource(&MBQuery).into(),
//Tool::from_datasource(&BandcampSource).into(),
Tool::from_datasource(&BeetsDB).into(),
Tool::from_datasource(&MixxxDB).into(),
// TODO: We should be able to have eva update lore memories with a function call, and this lore is somehow fed into the show? but only the relevant bits? or maybe eva even queries it directly
// TODO: The memory should also be able to remember facts about artists, albums, tracks we've had in the past, and those could be pulled up when there are hits in the playlist.
]
}
async fn execute_tool(&mut self, call: &ChatCompletionMessageToolCall) -> Result<ToolResults, ToolError> {
let func_name = call.function.name.as_str();
let args = call.function.arguments.as_str();
match func_name {
"query_bandcamp" => Ok(ToolResults { result: None, messages: vec![] }),
"query_beets" => self.tool_artifact_query(&mut BeetsDB, args).await,
"query_musicbrainz" => self.tool_artifact_query(&mut MBQuery, args).await,
"query_mixxx" => self.tool_artifact_query(&mut MixxxDB, args).await,
"synchronize_artifacts" => self.synchronize_artifacts().await,
"task_list" => self.tasklist_operation(args).await,
_ => Err(ToolError::InvalidToolName)
}
}
}
impl ArchiveToolbox {
async fn tasklist_operation(&mut self, json_args: &str) -> Result<ToolResults, ToolError> {
let args: TaskListArgs = serde_json::from_str(json_args)?;
let mut locked = self.todo_list.lock().await;
for op in args.operations {
match op {
TaskListOperation::Add(task) => {
locked.insert(task, false);
},
TaskListOperation::Complete(task) => {
// FIXME: The computer seems to waste a lot of time marking already completed tasks as completed
if let Some(result) = locked.get_mut(&task) {
*result = true;
}
},
}
}
Ok(ToolResults {
..Default::default()
})
}
async fn synchronize_artifacts(&mut self) -> Result<ToolResults, ToolError> {
let updated_count = self.archive.lock().await.synchronize().await;
Ok(ToolResults {
messages: vec![ConversationEntry::Spoken(Speaker::ShipComputer, format!("Synchronized {} items", updated_count))],
..Default::default()
})
}
async fn tool_artifact_query<Src: DataSource>(&mut self, src: &mut Src, json_args: &str) -> Result<ToolResults, ToolError> where Src::Args: core::fmt::Debug, Src::Error: core::fmt::Debug {
let args: Src::Args = serde_json::from_str(json_args)?;
log::debug!("Executing query {:?}", args);
let result;
match src.query(&args).await {
Ok(output) => {
result = format!("Found {} artifacts with archive query {:?}", output.len(), args);
for result in output {
self.archive.lock().await.insert(result);
}
},
Err(err) => {
result = format!("Unable to execute query: {:?}", err);
}
}
Ok(ToolResults {
result: Some(result),
messages: vec![]
})
}
}
#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
enum StageEvent {
ShipComputer(String),
StageDirection(String)
}
#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
struct StageEventArgs {
event: StageEvent
}
#[derive(Default, Debug)]
pub struct ToolResults {
pub result: Option<String>,
pub messages: Vec<ConversationEntry>
}
+35 -23
View File
@@ -1,34 +1,44 @@
use std::fmt::Display;
use async_openai::types::chat::{ChatCompletionRequestAssistantMessage, ChatCompletionRequestAssistantMessageContent, ChatCompletionRequestMessage, ChatCompletionRequestSystemMessage, ChatCompletionRequestSystemMessageContent, ChatCompletionRequestUserMessage, ChatCompletionRequestUserMessageContent};
use ratatui::style::{self, Style};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash, Copy)]
pub enum Speaker {
User,
ShipComputer,
Eva
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConversationEntry {
User(String),
Eva(String),
ShipComputer(String),
Spoken(Speaker, String),
StageDirection(String),
#[serde(skip)]
SystemMessage(String)
SystemMessage(String),
ShipComputerCommand(String)
}
impl ConversationEntry {
pub fn prefix(&self) -> Option<&str> {
match self {
ConversationEntry::Eva(_) => Some("Eva: "),
ConversationEntry::User(_) => Some("Argee: "),
ConversationEntry::ShipComputer(_) => Some("Ship Computer: "),
ConversationEntry::Spoken(Speaker::Eva, _) => Some("Eva: "),
ConversationEntry::Spoken(Speaker::User, _) => Some("Argee: "),
ConversationEntry::Spoken(Speaker::ShipComputer, _) => Some("Ship Computer: "),
ConversationEntry::ShipComputerCommand(_) => Some("> "),
_ => None,
}
}
pub fn prefix_style(&self) -> Style {
match self {
ConversationEntry::Eva(_) => Style::new().fg(style::Color::Cyan),
ConversationEntry::User(_) => Style::new().fg(style::Color::Magenta),
ConversationEntry::ShipComputer(_) => Style::new().fg(style::Color::Red),
ConversationEntry::Spoken(Speaker::Eva, _) => Style::new().fg(style::Color::Cyan),
ConversationEntry::Spoken(Speaker::User, _) => Style::new().fg(style::Color::Magenta),
ConversationEntry::Spoken(Speaker::ShipComputer, _) => Style::new().fg(style::Color::Red),
ConversationEntry::StageDirection(_) => Style::new().fg(style::Color::Yellow),
ConversationEntry::SystemMessage(_) => Style::new().fg(style::Color::DarkGray),
ConversationEntry::ShipComputerCommand(_) => Style::new().fg(style::Color::Red).bold(),
}
}
@@ -36,29 +46,31 @@ impl ConversationEntry {
match self {
ConversationEntry::StageDirection(_) => Style::new().fg(style::Color::Yellow),
ConversationEntry::SystemMessage(_) => Style::new().fg(style::Color::DarkGray),
ConversationEntry::ShipComputerCommand(_) => Style::new().fg(style::Color::Red).italic(),
_ => Style::new()
}
}
}
impl ToString for ConversationEntry {
fn to_string(&self) -> String {
match self {
ConversationEntry::Eva(text) => text,
ConversationEntry::ShipComputer(text) => text,
impl Display for ConversationEntry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
ConversationEntry::Spoken(Speaker::Eva, text) => text,
ConversationEntry::Spoken(Speaker::ShipComputer, text) => text,
ConversationEntry::StageDirection(text) => text,
ConversationEntry::SystemMessage(text) => text,
ConversationEntry::User(text) => text
}.clone()
ConversationEntry::Spoken(Speaker::User, text) => text,
ConversationEntry::ShipComputerCommand(text) => text
})
}
}
impl TryInto<ChatCompletionRequestMessage> for ConversationEntry {
fn try_into(self) -> Result<ChatCompletionRequestMessage, Self::Error> {
match self {
ConversationEntry::User(text) => Ok(ChatCompletionRequestMessage::User(ChatCompletionRequestUserMessage { content: text.into(), ..Default::default()})),
ConversationEntry::Eva(text) => Ok(ChatCompletionRequestMessage::Assistant(ChatCompletionRequestAssistantMessage { content: Some(text.into()), ..Default::default()})),
ConversationEntry::ShipComputer(text) => Ok(ChatCompletionRequestMessage::System(ChatCompletionRequestSystemMessage { content: text.into(), name: Some("ship-computer".into()), ..Default::default() })),
ConversationEntry::Spoken(Speaker::User, text) => Ok(ChatCompletionRequestMessage::User(ChatCompletionRequestUserMessage { content: text.into(), ..Default::default()})),
ConversationEntry::Spoken(Speaker::Eva, text) => Ok(ChatCompletionRequestMessage::Assistant(ChatCompletionRequestAssistantMessage { content: Some(text.into()), ..Default::default()})),
ConversationEntry::Spoken(Speaker::ShipComputer, text) => Ok(ChatCompletionRequestMessage::System(ChatCompletionRequestSystemMessage { content: text.into(), name: Some("ship-computer".into()), ..Default::default() })),
ConversationEntry::StageDirection(text) => Ok(ChatCompletionRequestMessage::System(ChatCompletionRequestSystemMessage { content: text.into(), name: Some("stage-direction".into()), ..Default::default() })),
_ => Err(())
}
@@ -71,11 +83,11 @@ impl TryInto<ChatCompletionRequestMessage> for ConversationEntry {
impl TryInto<ConversationEntry> for ChatCompletionRequestMessage {
fn try_into(self) -> Result<ConversationEntry, Self::Error> {
match self {
ChatCompletionRequestMessage::User(ChatCompletionRequestUserMessage { content: ChatCompletionRequestUserMessageContent::Text(msg), ..}) => Ok(ConversationEntry::User(msg)),
ChatCompletionRequestMessage::Assistant(ChatCompletionRequestAssistantMessage { content: Some(ChatCompletionRequestAssistantMessageContent::Text(msg)), ..}) => Ok(ConversationEntry::Eva(msg)),
ChatCompletionRequestMessage::User(ChatCompletionRequestUserMessage { content: ChatCompletionRequestUserMessageContent::Text(msg), ..}) => Ok(ConversationEntry::Spoken(Speaker::User, msg)),
ChatCompletionRequestMessage::Assistant(ChatCompletionRequestAssistantMessage { content: Some(ChatCompletionRequestAssistantMessageContent::Text(msg)), ..}) => Ok(ConversationEntry::Spoken(Speaker::Eva, msg)),
ChatCompletionRequestMessage::System(ChatCompletionRequestSystemMessage { content: ChatCompletionRequestSystemMessageContent::Text(msg), name: Some(name), ..}) => {
match name.as_str() {
"ship-computer" => Ok(ConversationEntry::ShipComputer(msg)),
"ship-computer" => Ok(ConversationEntry::Spoken(Speaker::ShipComputer, msg)),
"stage-direction" => Ok(ConversationEntry::StageDirection(msg)),
_ => Err(())
}
+15 -25
View File
@@ -1,7 +1,9 @@
use std::collections::HashMap;
use chrono::{DateTime, Duration, Utc};
use serde::{Deserialize, Serialize};
use crate::{artifacts::Artifact, prediction::{GeneratedResponses, PossibleResponse}, scene::conversation::ConversationEntry};
use crate::artifacts::{Track, archive::Archive};
pub mod conversation;
@@ -34,29 +36,25 @@ impl Default for StageDirection {
}
}
#[derive(Debug, Default, Serialize, Deserialize, Clone)]
pub struct Scenery {
pub artifacts: Vec<Artifact>,
pub current_playlist: Vec<Artifact>
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Scene {
reply_options: GeneratedResponses,
conversation: Vec<ConversationEntry>,
direction: StageDirection,
pub tokens_consumed: usize,
scenery: Scenery
pub current_playlist: Vec<Track>,
pub artifact_count: usize,
pub artifact_stats: (usize, usize, usize),
pub computer_task_list: HashMap<String, bool>
}
impl Scene {
pub fn new(reply_options: GeneratedResponses, conversation: Vec<ConversationEntry>, scenery: Scenery, tokens_consumed: usize, direction: StageDirection) -> Self {
pub fn new(tokens_consumed: usize, direction: StageDirection, archive: &Archive, current_playlist: Vec<Track>, computer_task_list: HashMap<String, bool>) -> Self {
Self {
reply_options,
conversation,
scenery,
tokens_consumed,
direction
direction,
current_playlist,
artifact_count: archive.len(),
artifact_stats: archive.stats(),
computer_task_list
}
}
@@ -64,15 +62,7 @@ impl Scene {
&self.direction
}
pub fn scenery(&self) -> &Scenery {
&self.scenery
}
pub fn conversation(&self) -> &Vec<ConversationEntry> {
&self.conversation
}
pub fn reply_options(&self) -> &Vec<PossibleResponse> {
&self.reply_options.responses
pub fn playlist(&self) -> &Vec<Track> {
&self.current_playlist
}
}
+112
View File
@@ -0,0 +1,112 @@
use rand::seq::IteratorRandom;
use symphonia::core::{formats::{TrackType, probe::Hint}, io::MediaSourceStream};
use crate::audio::AudioOutStream;
#[derive(Debug)]
pub enum SfxRequest {
RandomAmbient
}
#[derive(Debug)]
pub struct SfxControl {
sink: tokio::sync::mpsc::Sender<SfxRequest>
}
impl SfxControl {
pub async fn play_ambient(&mut self) -> Result<(), tokio::sync::mpsc::error::SendError<SfxRequest>> {
self.sink.send(SfxRequest::RandomAmbient).await
}
}
pub async fn start_sfx(audio_sink: AudioOutStream) -> SfxControl {
let (event_sink, mut event_src) = tokio::sync::mpsc::channel(32);
tokio::spawn(async move {
let sfx_dir = std::path::Path::new("./sfx");
loop {
while let Some(event) = event_src.recv().await {
match event {
SfxRequest::RandomAmbient => {
let avail_files = std::fs::read_dir(sfx_dir).unwrap();
let chosen_file = avail_files.choose(&mut rand::rng()).unwrap().unwrap();
log::debug!("Queuing ambient sound playback with {:?}", chosen_file);
let sfx_fd = std::fs::File::open(chosen_file.path()).unwrap();
let mss = MediaSourceStream::new(Box::new(sfx_fd), Default::default());
let meta_opts = Default::default();
let fmt_opts = Default::default();
let mut hint = Hint::new();
hint.with_extension(".mp3");
let mut format = symphonia::default::get_probe()
.probe(&hint, mss, fmt_opts, meta_opts)
.expect("Unsupported audio format");
let track = format.default_track(TrackType::Audio).expect("No audio track");
let track_id = track.id;
let dec_opts = Default::default();
let mut decoder = symphonia::default::get_codecs()
.make_audio_decoder(
track.codec_params.as_ref().expect("codec params missing").audio().unwrap(),
&dec_opts
).expect("Unsupported audio codec");
let sample_rate = decoder.codec_params().sample_rate.unwrap();
let channel_num = decoder.codec_params().channels.as_ref().unwrap().count();
log::debug!("Resampling {} -> {}", sample_rate, audio_sink.sample_rate);
// Our resampler works on a mono input
let mut bitrate_resample = resampler::ResamplerFir::new_from_hz(channel_num, sample_rate, audio_sink.sample_rate, Default::default(), Default::default());
log::debug!("Starting stream");
let mut audio_out_buf = vec![];
let mut channel_bufs: Vec<f32> = vec![];
loop {
let packet = match format.next_packet() {
Ok(Some(packet)) => packet,
Ok(None) => break,
Err(err) => panic!()
};
if packet.track_id != track_id {
continue
}
match decoder.decode_ref(&packet.as_packet_ref()) {
Ok(samples) => {
channel_bufs.resize(samples.samples_interleaved(), 0.);
samples.copy_to_slice_interleaved(&mut channel_bufs);
let mut resampled = [0.; 2048];
let (_, write_count) = bitrate_resample.resample(&channel_bufs, &mut resampled).unwrap();
// First we convert the audio feed from stereo down to mono by simple average
// TODO: This should be something smarter, like a saturating add..?
let mono_stream = resampled[..write_count].chunks(channel_num).map(|channels| {
let total_volume = channels.iter().cloned().reduce(|a, b| a + b).unwrap_or_default();
total_volume / (channel_num as f32)
});
// Then we write out the resampled audio to our staging buffer
audio_out_buf.extend(mono_stream);
// Once we have 1024 samples (jack default, I guess), we send it to the audio output
if audio_out_buf.len() >= 1024 {
audio_sink.sink.send(audio_out_buf).await.unwrap();
audio_out_buf = vec![];
}
},
Err(err) => panic!()
}
}
if !audio_out_buf.is_empty() {
audio_sink.sink.send(audio_out_buf).await.unwrap();
}
log::debug!("Playback complete");
}
}
}
}
});
SfxControl {
sink: event_sink
}
}
+12 -9
View File
@@ -14,19 +14,22 @@ You are playing the role of the artificial intelligence in a spaceship computer.
Along the way, you have the opportunity to invent lore and backstory for yourself, Argee, and the ship you both inhabit. This lore will be developed over the season.
To support your roleplaying, you have access to a sizable music library via the "archive_query" tool function. Internally, this runs `beet export` to produce json output.
To support your roleplaying, you have access to a sizable shipboard music archive.
You will occasionally be asked by Argee for information on the contents of the archive and how they are related to tracks in the playlist.
You also may use the archive to decide whether or not an artifact is somehow "familiar" based on whether or not it can be found there.
You must provide at least one parameter when calling this tool; it is wasteful to call it without any arguments.
The archive can be manipulated by interacting with the ship computer through the "stage_event" tool.
When interacting with it, you should use straightforward english commands or questions without roleplay.
For example, to have the computer search for information regarding the artist "nullsleep", you would tell the computer:
Another tool function named "musicbrainz_track_search" can be given a list of musicbrainz IDs (mbids), which will substantially expand the information available in the ship's artifact library.
You are able to use this function whenever it might be helpful to look up missing albums, tracks, or artists.
You should immediately run this tool against any new or unfamiliar musicbrainz IDs for tracks that get added to the list of artifacts available.
"load artist nullsleep"
There also exists a "bandcamp_artifact_scan" tool function, which will execute a search on Bandcamp and return a JSON list of artists and albums matching the query.
You are able to run multiple queries in parallel, and it is expected that you will run this tool whenever there is something unfamiliar in the playlist for the current episode, or Argee asks you for more information about the items in the playlist.
Eventually it will inform you when it has completed the tasks.
Queries to bandcamp are somewhat expensive, so you should check with the archive and musicbrainz first if you haven't already.
While it is working on its tasks, it will occasionally announce its current progress.
These messages should only be used as an indication of what the computer is doing.
You should only be replying to the computer's output if it is required to complete the task that Argee has given you.
You can ask it questions about items in the archive and it will attempt to fetch what it can from the outside world.
If you are asked to load information about the archive, or somehow find data is missing from the available list of artifacts, you must ask the ship computer to load the required data.
# Scene
The show features Argee, the main character of the show.
+8 -8
View File
@@ -4,7 +4,7 @@ use async_openai::{Client, config::OpenAIConfig, types::{InputSource, audio::{Au
use tempfile::SpooledData;
use tokio::sync::{mpsc, watch};
use crate::{audio::MicStream, events::AudioRecordRequest};
use crate::{audio::{AudioError, AudioInStream}, events::AudioRecordRequest};
#[derive(Debug)]
pub struct TranscriptionControl {
@@ -13,16 +13,16 @@ pub struct TranscriptionControl {
}
impl TranscriptionControl {
pub fn start(&mut self) {
self.record_state_sink.send(AudioRecordRequest::Start).unwrap();
pub fn start(&mut self) -> Result<(), AudioError> {
Ok(self.record_state_sink.send(AudioRecordRequest::Start)?)
}
pub fn stop(&mut self) {
self.record_state_sink.send(AudioRecordRequest::Finish).unwrap();
pub fn stop(&mut self) -> Result<(), AudioError> {
Ok(self.record_state_sink.send(AudioRecordRequest::Finish)?)
}
pub async fn next(&mut self) -> String {
self.transcription_result_src.recv().await.unwrap()
pub async fn next(&mut self) -> Option<String> {
self.transcription_result_src.recv().await
}
}
@@ -44,7 +44,7 @@ impl<T: std::io::Seek> std::io::Seek for RcFile<T> {
}
}
pub async fn start_transcription(mut mic_src: MicStream) -> TranscriptionControl {
pub async fn start_transcription(mut mic_src: AudioInStream) -> TranscriptionControl {
let (audio_control_in, mut audio_control_out) = watch::channel(AudioRecordRequest::Finish);
let (transcription_in, transcription_out) = mpsc::channel(1);
+4 -4
View File
@@ -1,6 +1,6 @@
use std::process::{Command, Stdio};
use crate::audio::TtsOutStream;
use crate::audio::AudioOutStream;
#[derive(Debug)]
pub struct TtsControl {
@@ -8,12 +8,12 @@ pub struct TtsControl {
}
impl TtsControl {
pub async fn speak(&self, text: String) {
self.request_sink.send(text).await.unwrap();
pub async fn speak(&self, text: String) -> Result<(), tokio::sync::mpsc::error::SendError<String>> {
self.request_sink.send(text).await
}
}
pub async fn start_tts(audio_sink: TtsOutStream) -> TtsControl {
pub async fn start_tts(audio_sink: AudioOutStream) -> TtsControl {
let (tts_request_sender, mut tts_request_receiver) = tokio::sync::mpsc::channel(3);
+58 -31
View File
@@ -1,23 +1,26 @@
use chrono::{Duration, Utc};
use crossterm::event::{self, KeyCode, KeyModifiers};
use ratatui::{Frame, layout::{Direction, Layout, Rect}, style::{self, }, text::Span, widgets::{BorderType, Clear, ListState, Paragraph}};
use ratatui::{Frame, layout::{Direction, Layout, Rect}, style::{self, Style, Stylize, }, text::Span, widgets::{BorderType, Clear, List, ListState, Paragraph}};
use throbber_widgets_tui::{Throbber, ThrobberState};
use tokio::time::Instant;
use tui_input::{Input, backend::crossterm::EventHandler};
use tui_skeleton::{AnimationMode, Block, Constraint, SkeletonText};
use crate::{audio::AudioInputControl, prediction::{PredictionAction, SessionControl, SessionUpdate}, scene::{Scene, conversation::ConversationEntry}, transcription::TranscriptionControl, tts::TtsControl};
use crate::{audio::AudioInputControl, prediction::{GeneratedResponses, PredictionAction, SessionControl, SessionUpdate}, scene::{Scene, conversation::{ConversationEntry, Speaker}}, transcription::TranscriptionControl, tts::TtsControl};
use crate::widgets::*;
#[derive(Debug)]
pub struct Ui {
scene: Scene,
reply_options: GeneratedResponses,
reply_state: ListState,
conversation_state: ListState,
user_input: Input,
throbber_state: ThrobberState,
is_requesting: bool,
computer_is_thinking: bool,
audio_level: f64,
recording_audio: bool,
focus_state: FocusState,
@@ -26,7 +29,8 @@ pub struct Ui {
transcription: TranscriptionControl,
audio: AudioInputControl,
tts: TtsControl,
predictions: SessionControl
predictions: SessionControl,
conversation: Vec<ConversationEntry>
}
#[derive(Debug)]
@@ -44,6 +48,7 @@ impl Ui {
user_input: Default::default(),
throbber_state: Default::default(),
is_requesting: false,
computer_is_thinking: false,
audio_level: -60.,
audio,
recording_audio: false,
@@ -51,7 +56,9 @@ impl Ui {
focus_state: FocusState::UserInput,
tts,
predictions,
last_tick: Instant::now()
last_tick: Instant::now(),
conversation: vec![],
reply_options: Default::default()
}
}
@@ -65,7 +72,7 @@ impl Ui {
.block(borders);
frame.render_widget(list, area);
} else {
frame.render_stateful_widget(Options(self.scene.reply_options()), area, &mut self.reply_state);
frame.render_stateful_widget(Options(&self.reply_options.responses), area, &mut self.reply_state);
}
}
@@ -81,10 +88,20 @@ impl Ui {
fn draw_io_throbber(&mut self, frame: &mut Frame, area: Rect) {
let throb_area = area.centered(Constraint::Max(1), Constraint::Max(1));
if self.is_requesting {
let throbber = Throbber::default();
let throbber = Throbber::default().throbber_style(Style::new().cyan());
frame.render_stateful_widget(throbber, throb_area, &mut self.throbber_state);
} else {
frame.render_widget(Clear::default(), throb_area);
frame.render_widget(Clear, throb_area);
}
}
fn draw_computer_throbber(&mut self, frame: &mut Frame, area: Rect) {
let throb_area = area.centered(Constraint::Min(1), Constraint::Min(1));
if self.computer_is_thinking {
let throbber = Throbber::default().throbber_style(Style::new().red()).throbber_set(throbber_widgets_tui::WHITE_SQUARE);
frame.render_stateful_widget(throbber, throb_area, &mut self.throbber_state);
} else {
frame.render_widget(Clear, throb_area);
}
}
@@ -111,31 +128,34 @@ impl Ui {
let scene_layout = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Fill(4), Constraint::Fill(1)])
.constraints([Constraint::Fill(5), Constraint::Fill(2)])
.split(layout[0]);
frame.render_stateful_widget(Conversation(self.scene.conversation()), scene_layout[0], &mut self.conversation_state);
self.draw_narration(frame, scene_layout[1]);
frame.render_stateful_widget(Conversation(&self.conversation), scene_layout[0], &mut self.conversation_state);
//self.draw_narration(frame, scene_layout[1]);
//self.draw_tasklist(frame, scene_layout[1]);
frame.render_widget(TaskList(&self.scene.computer_task_list), scene_layout[1]);
self.draw_options(frame, layout[1]);
let status_layout = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Max(3), Constraint::Fill(2), Constraint::Max(13), Constraint::Min(50)])
.constraints([Constraint::Max(3), Constraint::Max(3), Constraint::Fill(2), Constraint::Max(13), Constraint::Min(50)])
.split(layout[3]);
self.draw_user_input(frame, layout[2]);
self.draw_io_throbber(frame, status_layout[0]);
frame.render_widget(StatusBar(&self.scene), status_layout[1]);
frame.render_widget(RecordingStatus(self.recording_audio), status_layout[2]);
frame.render_widget(Volume(self.audio_level, self.recording_audio), status_layout[3]);
self.draw_computer_throbber(frame, status_layout[1]);
frame.render_widget(StatusBar(&self.scene), status_layout[2]);
frame.render_widget(RecordingStatus(self.recording_audio), status_layout[3]);
frame.render_widget(Volume(self.audio_level, self.recording_audio), status_layout[4]);
}
async fn insert_selected_prompt(&mut self) {
let selected = self.scene.reply_options()[self.reply_state.selected().unwrap()].clone();
let selected = self.reply_options.responses[self.reply_state.selected().unwrap()].clone();
if let Some(direction) = &selected.stage_direction {
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::StageDirection(direction.clone()))).await;
}
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Eva(selected.text.clone()))).await;
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Spoken(Speaker::Eva, selected.text.clone()))).await;
}
async fn on_command(&mut self, command: &str) {
@@ -150,7 +170,6 @@ impl Ui {
self.predictions.insert(PredictionAction::SetPlaylist(playlist_name)).await;
} else {
log::error!("Invalid episode number format. Use /episode [number]");
return;
}
},
"/playlist" => {
@@ -172,7 +191,7 @@ impl Ui {
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::StageDirection(arg.to_string()))).await;
},
"/computer" => {
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::ShipComputer(arg.to_string()))).await;
self.predictions.insert(PredictionAction::ComputerCommand(arg.to_string())).await;
},
_ => {
log::error!("Unknown command. Available commands: /episode [number], /narrative [text], /event [text], /computer [text], /timer [minutes]");
@@ -181,6 +200,7 @@ impl Ui {
}
pub async fn on_event(&mut self, evt: event::Event) {
// TODO: ctrl+l should drop all the SystemMessage entries from the log to clean up the UI
if let Some(key) = evt.as_key_press_event() {
match self.focus_state {
FocusState::Conversation => {
@@ -198,8 +218,8 @@ impl Ui {
KeyCode::Up => self.conversation_state.select_next(),
KeyCode::Enter => {
let row_num = self.conversation_state.selected().unwrap();
if let ConversationEntry::Eva(text) = &self.scene.conversation()[self.scene.conversation().len() - 1 - row_num] {
self.tts.speak(text.clone()).await;
if let ConversationEntry::Spoken(Speaker::Eva, text) = &self.conversation[self.conversation.len() - 1 - row_num] {
self.tts.speak(text.clone()).await.unwrap();
self.focus_state = FocusState::UserInput;
self.conversation_state.select(None);
self.reply_state.select_first();
@@ -219,10 +239,10 @@ impl Ui {
KeyCode::Char('x') if key.modifiers.contains(KeyModifiers::CONTROL) => {
if self.recording_audio {
self.recording_audio = false;
self.transcription.stop();
self.transcription.stop().unwrap();
} else {
self.recording_audio = true;
self.transcription.start();
self.transcription.start().unwrap();
}
},
KeyCode::Down => self.reply_state.select_next(),
@@ -234,12 +254,10 @@ impl Ui {
let next_msg = self.user_input.value_and_reset();
if next_msg.trim().is_empty() {
self.insert_selected_prompt().await;
} else if next_msg.starts_with("/") {
self.on_command(&next_msg).await;
} else {
if next_msg.starts_with("/") {
self.on_command(&next_msg).await;
} else {
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::User(next_msg))).await;
}
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Spoken(Speaker::User, next_msg))).await;
}
},
_ => {self.user_input.handle_event(&evt);},
@@ -257,21 +275,30 @@ impl Ui {
}
tokio::select!{
_ = tokio::time::sleep(std::time::Duration::from_millis(60)), if self.is_requesting => (),
_ = tokio::time::sleep(std::time::Duration::from_millis(60)), if self.is_requesting || self.computer_is_thinking => (),
next_update = self.predictions.changed() => {
match next_update {
SessionUpdate::Thinking(is_thinking) => self.is_requesting = is_thinking,
SessionUpdate::Thinking(Speaker::Eva, is_thinking) => self.is_requesting = is_thinking,
SessionUpdate::Thinking(Speaker::ShipComputer, is_thinking) => self.computer_is_thinking = is_thinking,
SessionUpdate::Thinking(_, _) => unreachable!(),
SessionUpdate::Scene(scene) => {
self.scene = scene;
self.reply_state.select_first();
},
SessionUpdate::Responses(responses) => {
self.reply_options = responses;
self.reply_state.select_first();
}
SessionUpdate::Conversation(conversation) => {
self.conversation = conversation;
},
}
},
next_volume = self.audio.next() => {
self.audio_level = next_volume
self.audio_level = next_volume.expect("Audio volume event stream has died")
},
transcription_result = self.transcription.next() => {
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::User(transcription_result))).await;
self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Spoken(Speaker::User, transcription_result.expect("Transcription stream has died")))).await;
},
}
}
+43 -9
View File
@@ -1,3 +1,5 @@
use std::collections::HashMap;
use chrono::Duration;
use ratatui::{layout::Rect, style::Style, text::{Line, Text}, widgets::*};
use tui_skeleton::Block;
@@ -5,6 +7,33 @@ use ratatui::prelude::*;
use crate::{prediction::PossibleResponse, scene::{Scene, conversation::ConversationEntry}};
pub struct TaskList<'a>(pub &'a HashMap<String, bool>);
impl Widget for TaskList<'_> {
fn render(self, area: Rect, buf: &mut Buffer)
where
Self: Sized {
let borders = Block::bordered().border_style(style::Color::Red).title("Computer Tasks");
let wrap_options_unfinished = textwrap::Options::new(area.width as usize).initial_indent("[ ] ").subsequent_indent(" ");
let wrap_options_finished = textwrap::Options::new(area.width as usize).initial_indent("[X] ").subsequent_indent(" ");
let options: Vec<Text> = self.0.iter().map(|(text, is_finished)| {
let (options, color) = if *is_finished {
(wrap_options_finished.clone(), style::Color::DarkGray)
} else {
(wrap_options_unfinished.clone(), style::Color::Green)
};
let contents: Vec<Line> = textwrap::wrap(text, options)
.iter()
.map(|x| { Line::from(x.to_string()).fg(color)}).collect();
Text::from_iter(contents)
}).collect();
let list = List::new(options)
.block(borders)
.style(ratatui::style::Color::White);
Widget::render(list, area, buf);
}
}
pub struct Options<'a>(pub &'a Vec<PossibleResponse>);
impl StatefulWidget for Options<'_> {
@@ -17,16 +46,18 @@ impl StatefulWidget for Options<'_> {
if let Some(direction) = &option.stage_direction {
let padded = format!("({})", direction);
let mut wrapped_direction: Vec<Line> = textwrap::wrap(&padded, wrap_options.clone())
let wrapped = textwrap::wrap(&padded, wrap_options.clone());
let wrapped_direction = wrapped
.iter()
.map(|x| { Line::from(x.to_string()).fg(style::Color::Yellow)}).collect();
contents.append(&mut wrapped_direction);
.map(|x| { Line::from(x.to_string()).fg(style::Color::Yellow)});
contents.extend(wrapped_direction);
}
let mut text: Vec<Line> = textwrap::wrap(&option.text, wrap_options.clone())
let wrapped = textwrap::wrap(&option.text, wrap_options.clone());
let text = wrapped
.iter()
.map(|x| { Line::from(x.to_string())}).collect();
contents.append(&mut text);
.map(|x| { Line::from(x.to_string())});
contents.extend(text);
Text::from_iter(contents)
}).collect();
let list = List::new(options)
@@ -153,7 +184,7 @@ impl Widget for StatusBar<'_> {
let negative = time_remaining.abs() != time_remaining;
let time_style = if minutes_remaining <= 0 || negative {
Style::new().fg(ratatui::style::Color::LightRed).bold()
Style::new().fg(ratatui::style::Color::LightRed).underlined()
} else if minutes_remaining < 5 {
Style::new().fg(ratatui::style::Color::LightRed).bold()
} else if minutes_remaining < 10 {
@@ -172,14 +203,17 @@ impl Widget for StatusBar<'_> {
format!("{:0>2}:{:0>2}:{:0>2}", time_remaining.num_hours(), time_remaining.num_minutes() % 60, time_remaining.num_seconds() % 60)
};
let artifact_count = self.0.artifact_count;
let artifact_stats = self.0.artifact_stats;
let status_line = Line::from_iter([
Span::from(format!("Playlist: {}", self.0.direction().playlist)).style(ratatui::style::Color::LightBlue),
Span::from(" | ").style(ratatui::style::Color::DarkGray),
Span::from(format!("{} tracks", self.0.scenery().current_playlist.len())).style(ratatui::style::Color::LightBlue),
Span::from(format!("{} tracks", self.0.playlist().len())).style(ratatui::style::Color::LightBlue),
Span::from(" | ").style(ratatui::style::Color::DarkGray),
Span::from(format!("Time Remaining: {}", formatted_time)).style(time_style),
Span::from(" | ").style(ratatui::style::Color::DarkGray),
Span::from(format!("{} artifacts recorded", self.0.scenery().artifacts.len())).style(ratatui::style::Color::LightBlue),
Span::from(format!("{} ({}/{}/{}) artifacts recorded", artifact_count, artifact_stats.0, artifact_stats.1, artifact_stats.2)).style(ratatui::style::Color::LightBlue),
Span::from(" | ").style(ratatui::style::Color::DarkGray),
Span::from(format!("{} tokens sacrificed", self.0.tokens_consumed)).style(ratatui::style::Color::LightCyan),
// TODO: Should show the available and consumed context window in terms of tokens here