pub use kcode_k1_audio_classification_projection::{
ExecutedAnalysis, FragmentId, FragmentStageV1, FragmentStatus, LlmJobState, LlmJobStatus,
OverallState, SpeakerLabelV1, StageState, StageStatus,
};
use kcode_k1_audio_classification_projection::{Projection, ProjectionEffect};
use kcode_k1_audio_fragment_runner as runner;
use kcode_k1_audio_fragment_submit as fragment_submit;
use kcode_k1_audio_fragment_transactions as transactions;
use kcode_k1_objects::K1Objects;
use kcode_k1_peering::K1Peering;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem, SubsystemId, TxId};
use kcode_speaker_v3_analysis::Analyzer;
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::Path;
#[cfg(test)]
use std::path::PathBuf;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use tokio::sync::mpsc;
use tokio::task::JoinHandle as LocalJoinHandle;
const RESTART_ERROR: &str = "analysis interrupted by restart";
const SUBSYSTEM_NAME: &str = "audio-classification";
pub struct AudioClassification {
projection: Arc<Projection>,
_ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
control: Arc<Control>,
health: Arc<Health>,
worker: Option<JoinHandle<()>>,
}
impl AudioClassification {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
analyzer: Analyzer,
) -> Result<Self, String> {
Self::open_with_engine(
root,
ordering,
peering,
objects,
Box::new(ProductionEngine { analyzer }),
)
}
fn open_with_engine(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
engine: Box<dyn Engine>,
) -> Result<Self, String> {
let (projection, cursor) = Projection::open(root, &ordering)?;
let projection = Arc::new(projection);
let control = Arc::new(Control::default());
let health = Arc::new(Health::default());
let replaying = Arc::new(AtomicBool::new(true));
let replayed = Arc::new(AtomicU64::new(0));
let callback = Arc::new(Callback {
projection: projection.clone(),
control: control.clone(),
health: health.clone(),
replaying: replaying.clone(),
replayed: replayed.clone(),
});
let (sender, receiver) = mpsc::unbounded_channel();
control.install(sender.clone());
let worker = spawn_worker(
engine,
sender,
receiver,
control.clone(),
health.clone(),
peering.clone(),
objects.clone(),
)?;
let startup = (|| -> Result<(), String> {
ordering
.register_subsystem(subsystem_id()?, cursor, callback)
.map_err(|error| format!("register audio classification: {error}"))?;
replaying.store(false, AtomicOrdering::Release);
record_replay(root, replayed.load(AtomicOrdering::Acquire));
let interrupted = projection
.running()
.map_err(|error| format!("read interrupted analyses: {error}"))?;
for fragment in interrupted {
transactions::submit_failure(
&peering,
fragment.fragment_id,
fragment.stage,
None,
RESTART_ERROR.to_owned(),
)
.map_err(|error| format!("persist interrupted analysis: {error}"))?;
}
control.activate();
let queued = projection
.queued()
.map_err(|error| format!("read queued analyses: {error}"))?;
for fragment_id in queued {
control.start(fragment_id, 0)?;
}
health.ensure()
})();
if let Err(error) = startup {
control.shutdown();
let _ = worker.join();
return Err(error);
}
Ok(Self {
projection,
_ordering: ordering,
peering,
objects,
control,
health,
worker: Some(worker),
})
}
pub fn submit(&self, ogg_bytes: &[u8]) -> Result<FragmentId, String> {
self.health.ensure()?;
fragment_submit::submit(&self.objects, &self.peering, ogg_bytes)
.map_err(|error| self.submission_error(error))
}
pub fn status(&self, fragment_id: FragmentId) -> Result<Option<FragmentStatus>, String> {
self.health.ensure()?;
self.projection.status(fragment_id)
}
pub fn retry(&self, fragment_id: FragmentId) -> Result<(), String> {
self.health.ensure()?;
let status = self
.projection
.status(fragment_id)?
.ok_or_else(|| "unknown audio fragment".to_owned())?;
if status.state != OverallState::Failed {
return Err("retry requires Failed state".to_owned());
}
if !self
.control
.start(fragment_id, status.attempt_count)
.map_err(|error| self.worker_error(error))?
{
return Err("retry is already active".to_owned());
}
Ok(())
}
pub fn discard(&self, fragment_id: FragmentId) -> Result<(), String> {
self.health.ensure()?;
let status = self
.projection
.status(fragment_id)?
.ok_or_else(|| "unknown audio fragment".to_owned())?;
if status.state == OverallState::Discarded {
return Ok(());
}
if !self.control.reserve_discard(fragment_id) {
return Err("discard is already active".to_owned());
}
match transactions::submit_discard(&self.peering, fragment_id) {
Ok(_) => {
self.control.discard_done(fragment_id);
Ok(())
}
Err(error) => {
if !is_committed_error(&error) {
self.control.discard_done(fragment_id);
}
Err(self.submission_error(error))
}
}
}
pub fn submit_labels(
&self,
fragment_id: FragmentId,
labels: Vec<SpeakerLabelV1>,
) -> Result<(), String> {
self.health.ensure()?;
let interim = self.projection.validate_labels(fragment_id, &labels)?;
if !self.control.reserve_labels(fragment_id) {
return Err("label submission is already active".to_owned());
}
match transactions::submit_label_confirmation(&self.peering, fragment_id, interim, labels) {
Ok(_) => {
self.control.labels_done(fragment_id);
Ok(())
}
Err(error) => {
if !is_committed_error(&error) {
self.control.labels_done(fragment_id);
}
Err(self.submission_error(error))
}
}
}
fn submission_error(&self, error: String) -> String {
if is_committed_error(&error) {
self.health
.fault("transaction commitment is ambiguous".to_owned());
self.control.shutdown();
}
error
}
fn worker_error(&self, error: String) -> String {
self.health.fault(error.clone());
self.control.shutdown();
error
}
#[cfg(test)]
fn inject_error_burst(&self, id: FragmentId, errors: Vec<String>) -> Result<(), String> {
self.projection.inject_errors(id, errors)
}
}
impl Drop for AudioClassification {
fn drop(&mut self) {
self.control.shutdown();
if let Some(worker) = self.worker.take() {
let _ = worker.join();
}
}
}
#[derive(Default)]
struct Health {
reopen: AtomicBool,
diagnostic: Mutex<Option<String>>,
}
impl Health {
fn ensure(&self) -> Result<(), String> {
if !self.reopen.load(AtomicOrdering::Acquire) {
return Ok(());
}
let detail = lock(&self.diagnostic)
.clone()
.unwrap_or_else(|| "processing fault".to_owned());
Err(format!("audio classification requires reopen: {detail}"))
}
fn fault(&self, diagnostic: String) {
let mut current = lock(&self.diagnostic);
if current.is_none() {
*current = Some(diagnostic);
}
self.reopen.store(true, AtomicOrdering::Release);
}
}
#[derive(Default)]
struct Control {
state: Mutex<ControlState>,
live: AtomicBool,
}
#[derive(Default)]
struct ControlState {
sender: Option<mpsc::UnboundedSender<WorkerCommand>>,
starting: HashMap<FragmentId, StartReservation>,
labels: HashSet<FragmentId>,
discards: HashSet<FragmentId>,
next_generation: u64,
}
struct StartReservation {
generation: u64,
baseline_attempt: u32,
}
impl Control {
fn install(&self, sender: mpsc::UnboundedSender<WorkerCommand>) {
lock(&self.state).sender = Some(sender);
}
fn activate(&self) {
self.live.store(true, AtomicOrdering::Release);
}
fn start(&self, id: FragmentId, baseline_attempt: u32) -> Result<bool, String> {
if !self.live.load(AtomicOrdering::Acquire) {
return Err("audio classification worker is unavailable".to_owned());
}
let mut state = lock(&self.state);
if state.starting.contains_key(&id) {
return Ok(false);
}
state.next_generation = state
.next_generation
.checked_add(1)
.ok_or_else(|| "audio classification worker generation overflow".to_owned())?;
let generation = state.next_generation;
let sender = state
.sender
.clone()
.ok_or_else(|| "audio classification worker is unavailable".to_owned())?;
state.starting.insert(
id,
StartReservation {
generation,
baseline_attempt,
},
);
drop(state);
if sender.send(WorkerCommand::Start(id, generation)).is_err() {
self.finished(id, generation);
return Err("audio classification worker is unavailable".to_owned());
}
Ok(true)
}
fn progress_applied(&self, id: FragmentId, attempt_count: u32) {
let mut state = lock(&self.state);
if state
.starting
.get(&id)
.is_some_and(|value| attempt_count > value.baseline_attempt)
{
state.starting.remove(&id);
}
}
fn abort(&self, id: FragmentId) -> Result<(), String> {
let sender = {
let mut state = lock(&self.state);
state.starting.remove(&id);
state.labels.remove(&id);
state.discards.remove(&id);
state.sender.clone()
};
if self.live.load(AtomicOrdering::Acquire)
&& sender.is_some_and(|sender| sender.send(WorkerCommand::Abort(id)).is_err())
{
return Err("audio classification worker is unavailable".to_owned());
}
Ok(())
}
fn finished(&self, id: FragmentId, generation: u64) {
let mut state = lock(&self.state);
if state
.starting
.get(&id)
.is_some_and(|value| value.generation == generation)
{
state.starting.remove(&id);
}
}
fn reserve_labels(&self, id: FragmentId) -> bool {
lock(&self.state).labels.insert(id)
}
fn labels_done(&self, id: FragmentId) {
lock(&self.state).labels.remove(&id);
}
fn reserve_discard(&self, id: FragmentId) -> bool {
lock(&self.state).discards.insert(id)
}
fn discard_done(&self, id: FragmentId) {
lock(&self.state).discards.remove(&id);
}
fn shutdown(&self) {
self.live.store(false, AtomicOrdering::Release);
let sender = lock(&self.state).sender.clone();
if let Some(sender) = sender {
let _ = sender.send(WorkerCommand::Stop);
}
}
}
struct Callback {
projection: Arc<Projection>,
control: Arc<Control>,
health: Arc<Health>,
replaying: Arc<AtomicBool>,
replayed: Arc<AtomicU64>,
}
impl Callback {
fn fault(&self, diagnostic: String) -> String {
self.health.fault(diagnostic.clone());
self.control.shutdown();
diagnostic
}
fn react(&self, fragment_id: FragmentId, effect: ProjectionEffect) -> Result<(), String> {
match effect {
ProjectionEffect::Start => {
if self.control.live.load(AtomicOrdering::Acquire) {
self.control.start(fragment_id, 0)?;
}
}
ProjectionEffect::Abort => self.control.abort(fragment_id)?,
ProjectionEffect::LabelsCommitted => self.control.labels_done(fragment_id),
ProjectionEffect::None => {
let status = self
.projection
.status(fragment_id)?
.ok_or_else(|| "applied event has no projected fragment".to_owned())?;
self.control
.progress_applied(fragment_id, status.attempt_count);
}
}
Ok(())
}
}
impl Subsystem for Callback {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
let applied = self
.projection
.apply(id, payload)
.map_err(|error| self.fault(format!("apply audio classification event: {error}")))?;
if self.replaying.load(AtomicOrdering::Acquire) {
self.replayed.fetch_add(1, AtomicOrdering::AcqRel);
return Ok(());
}
self.react(applied.fragment_id, applied.effect)
.map_err(|error| self.fault(format!("apply audio classification effect: {error}")))
}
fn reorg(&self) -> Result<(), String> {
let result = self.projection.clear();
self.health
.fault("canonical reorganization requires reopen".to_owned());
self.control.shutdown();
result
}
}
type EngineFuture<'a> = Pin<Box<dyn Future<Output = Result<(), String>> + 'a>>;
trait Engine: Send + 'static {
fn run<'a>(
&'a self,
peering: &'a K1Peering,
fragment_id: FragmentId,
ogg_bytes: &'a [u8],
) -> EngineFuture<'a>;
}
struct ProductionEngine {
analyzer: Analyzer,
}
impl Engine for ProductionEngine {
fn run<'a>(
&'a self,
peering: &'a K1Peering,
fragment_id: FragmentId,
ogg_bytes: &'a [u8],
) -> EngineFuture<'a> {
Box::pin(runner::run(&self.analyzer, peering, fragment_id, ogg_bytes))
}
}
enum WorkerCommand {
Start(FragmentId, u64),
Abort(FragmentId),
Finished(FragmentId, u64, Result<(), String>),
Stop,
}
fn spawn_worker(
engine: Box<dyn Engine>,
sender: mpsc::UnboundedSender<WorkerCommand>,
receiver: mpsc::UnboundedReceiver<WorkerCommand>,
control: Arc<Control>,
health: Arc<Health>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
) -> Result<JoinHandle<()>, String> {
thread::Builder::new()
.name("k1-audio-classification".to_owned())
.spawn(move || worker_main(engine, sender, receiver, control, health, peering, objects))
.map_err(|error| format!("start audio classification worker: {error}"))
}
fn worker_main(
engine: Box<dyn Engine>,
sender: mpsc::UnboundedSender<WorkerCommand>,
mut receiver: mpsc::UnboundedReceiver<WorkerCommand>,
control: Arc<Control>,
health: Arc<Health>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
) {
let runtime = match tokio::runtime::Builder::new_current_thread().build() {
Ok(runtime) => runtime,
Err(error) => {
health.fault(format!("create worker runtime: {error}"));
control.shutdown();
return;
}
};
let local = tokio::task::LocalSet::new();
runtime.block_on(local.run_until(async move {
let engine: Rc<dyn Engine> = Rc::from(engine);
let mut tasks: HashMap<FragmentId, (u64, LocalJoinHandle<()>)> = HashMap::new();
while let Some(command) = receiver.recv().await {
match command {
WorkerCommand::Start(id, generation) => {
if let Some((old_generation, old_task)) = tasks.remove(&id) {
old_task.abort();
control.finished(id, old_generation);
}
let task_engine = engine.clone();
let task_peering = peering.clone();
let task_objects = objects.clone();
let task_sender = sender.clone();
let task = tokio::task::spawn_local(async move {
let result =
run_fragment(task_engine.as_ref(), &task_peering, &task_objects, id)
.await;
let _ = task_sender.send(WorkerCommand::Finished(id, generation, result));
});
tasks.insert(id, (generation, task));
}
WorkerCommand::Abort(id) => {
if let Some((generation, task)) = tasks.remove(&id) {
task.abort();
control.finished(id, generation);
}
}
WorkerCommand::Finished(id, generation, result) => {
if tasks
.get(&id)
.is_some_and(|(current, _)| *current == generation)
{
tasks.remove(&id);
control.finished(id, generation);
}
if let Err(error) = result {
health.fault(format!("runner persistence failure: {error}"));
control.shutdown();
break;
}
}
WorkerCommand::Stop => break,
}
}
for (id, (generation, task)) in tasks {
task.abort();
control.finished(id, generation);
}
}));
}
async fn run_fragment(
engine: &dyn Engine,
peering: &K1Peering,
objects: &K1Objects,
id: FragmentId,
) -> Result<(), String> {
let object = match objects.load(id) {
Ok(Some(object)) if object.file_type == "audio/ogg" => object,
Ok(Some(_)) => {
transactions::submit_failure(
peering,
id,
FragmentStageV1::Queue,
None,
"audio Object is not audio/ogg".to_owned(),
)?;
return Ok(());
}
Ok(None) => {
transactions::submit_failure(
peering,
id,
FragmentStageV1::Queue,
None,
"audio Object is unavailable".to_owned(),
)?;
return Ok(());
}
Err(error) => {
transactions::submit_failure(
peering,
id,
FragmentStageV1::Queue,
None,
format!("load audio Object: {error}"),
)?;
return Ok(());
}
};
engine.run(peering, id, &object.data).await
}
fn subsystem_id() -> Result<SubsystemId, String> {
SubsystemId::from_str(SUBSYSTEM_NAME)
}
fn is_committed_error(error: &str) -> bool {
error.to_ascii_lowercase().contains("committed")
}
fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(|error| error.into_inner())
}
#[cfg(test)]
fn record_replay(root: &Path, count: u64) {
lock(replay_counts()).insert(root.to_path_buf(), count);
}
#[cfg(not(test))]
fn record_replay(_root: &Path, _count: u64) {}
#[cfg(test)]
fn replay_counts() -> &'static Mutex<HashMap<PathBuf, u64>> {
use std::sync::OnceLock;
static COUNTS: OnceLock<Mutex<HashMap<PathBuf, u64>>> = OnceLock::new();
COUNTS.get_or_init(|| Mutex::new(HashMap::new()))
}
#[cfg(test)]
mod tests {
use super::*;
use kcode_k1_audio_classification_testkit::{
AdapterError, AdapterFactory, ClassificationAdapter, OpenRequest,
SUCCESS_INTERIM_TRANSCRIPT, ScriptedOutcome, TestId, TestJob, TestJobState, TestState,
TestStatus, run_all,
};
use kcode_k1_audio_fragment_transactions::ProgressUpdateV1;
use kcode_k1_txn_ordering::REGISTER_AT_TIP;
use kcode_speaker_v3_analysis::{
AnalysisEnvelope, FeatureVector24, GeminiCohort, OggAudioMetadata, StructuredAnalysis,
StructuredSpeaker, StructurerProvenance,
};
use std::collections::VecDeque;
use tokio::sync::oneshot;
const PROJECTION_DATABASE_FILE: &str = "audio-classification.sqlite3";
struct ScriptedEngine {
outcomes: Mutex<VecDeque<ScriptedOutcome>>,
}
impl Engine for ScriptedEngine {
fn run<'a>(
&'a self,
peering: &'a K1Peering,
id: FragmentId,
bytes: &'a [u8],
) -> EngineFuture<'a> {
Box::pin(async move {
let Some(mut outcome) = lock(&self.outcomes).pop_front() else {
return Ok(());
};
transactions::submit_progress(
peering,
id,
ProgressUpdateV1::LlmJobStarted {
sequence: 1,
stage: FragmentStageV1::Transcript,
name: "transcript".to_owned(),
},
)?;
loop {
match outcome {
ScriptedOutcome::Blocked {
gate,
outcome: next,
} => {
let (sender, receiver) = oneshot::channel();
thread::Builder::new()
.name("audio-test-gate".to_owned())
.spawn(move || {
gate.wait_for_release();
let _ = sender.send(());
})
.map_err(|error| error.to_string())?;
receiver
.await
.map_err(|_| "test gate cancelled".to_owned())?;
outcome = *next;
}
ScriptedOutcome::Success => {
transactions::submit_progress(
peering,
id,
ProgressUpdateV1::LlmJobSucceeded { sequence: 1 },
)?;
transactions::submit_transcription_complete(
peering,
id,
success_analysis(bytes)?,
)?;
return Ok(());
}
ScriptedOutcome::Failure { message } => {
transactions::submit_failure(
peering,
id,
FragmentStageV1::Transcript,
Some(1),
message,
)?;
return Ok(());
}
}
}
})
}
}
fn success_analysis(bytes: &[u8]) -> Result<ExecutedAnalysis, String> {
let speaker_one = "Speaker 1".parse().map_err(|error| format!("{error}"))?;
let speaker_two = "Speaker 2".parse().map_err(|error| format!("{error}"))?;
let provenance = StructurerProvenance {
model_id: "test-model".to_owned(),
prompt_revision: "test-prompt".to_owned(),
};
Ok(ExecutedAnalysis {
envelope: AnalysisEnvelope {
audio: OggAudioMetadata::from_ogg_bytes(bytes)
.map_err(|error| error.to_string())?,
analysis: StructuredAnalysis {
transcript: SUCCESS_INTERIM_TRANSCRIPT.to_owned(),
speakers: vec![
StructuredSpeaker {
speaker: speaker_one,
language: "English".to_owned(),
features: FeatureVector24::default(),
features_usable_for_training: true,
},
StructuredSpeaker {
speaker: speaker_two,
language: "English".to_owned(),
features: FeatureVector24::default(),
features_usable_for_training: true,
},
],
},
gemini: GeminiCohort {
model_id: "test-gemini".to_owned(),
transcript_prompt_revision: "test-transcript".to_owned(),
feature_prompt_revisions: std::array::from_fn(|_| "test-feature".to_owned()),
feature_schema_revision: "test-schema".to_owned(),
},
structurer: provenance.clone(),
},
label_extractor: provenance,
})
}
struct TestFactory;
struct TestAdapter {
facade: AudioClassification,
}
impl AdapterFactory for TestFactory {
type Adapter = TestAdapter;
fn open(&self, request: OpenRequest) -> Result<Self::Adapter, AdapterError> {
let ordering = Arc::new(
K1TxnOrdering::open(&request.root.join("ordering")).map_err(adapter_error)?,
);
let peering = Arc::new(
K1Peering::open(&request.root.join("peering"), ordering.clone())
.map_err(adapter_error)?,
);
let objects = Arc::new(
K1Objects::open(ordering.clone(), peering.clone()).map_err(adapter_error)?,
);
let facade = AudioClassification::open_with_engine(
&request.root.join("projection"),
ordering,
peering,
objects,
Box::new(ScriptedEngine {
outcomes: Mutex::new(request.outcomes.into()),
}),
)
.map_err(adapter_error)?;
Ok(TestAdapter { facade })
}
fn append_queued_while_closed(
&self,
root: &Path,
bytes: &[u8],
) -> Result<TestId, AdapterError> {
let ordering =
Arc::new(K1TxnOrdering::open(&root.join("ordering")).map_err(adapter_error)?);
let peering = Arc::new(
K1Peering::open(&root.join("peering"), ordering.clone()).map_err(adapter_error)?,
);
let objects = Arc::new(
K1Objects::open(ordering.clone(), peering.clone()).map_err(adapter_error)?,
);
ordering
.register_subsystem(
subsystem_id().map_err(adapter_error)?,
Some(REGISTER_AT_TIP),
Arc::new(NoopSubsystem),
)
.map_err(adapter_error)?;
fragment_submit::submit(&objects, &peering, bytes)
.map(to_test_id)
.map_err(adapter_error)
}
fn corrupt_projection(&self, root: &Path) -> Result<(), AdapterError> {
std::fs::write(
root.join("projection").join(PROJECTION_DATABASE_FILE),
b"corrupt projection",
)
.map_err(|error| adapter_error(error.to_string()))
}
fn startup_replay_count(&self, root: &Path) -> Result<u64, AdapterError> {
lock(replay_counts())
.get(&root.join("projection"))
.copied()
.ok_or_else(|| adapter_error("startup replay count is unavailable".to_owned()))
}
}
impl ClassificationAdapter for TestAdapter {
fn submit(&self, bytes: &[u8]) -> Result<TestId, AdapterError> {
self.facade
.submit(bytes)
.map(to_test_id)
.map_err(adapter_error)
}
fn status(&self, id: TestId) -> Result<TestStatus, AdapterError> {
let status = self
.facade
.status(from_test_id(id))
.map_err(adapter_error)?
.ok_or_else(|| adapter_error("status is unavailable".to_owned()))?;
Ok(TestStatus {
state: match status.state {
OverallState::Queued => TestState::Queued,
OverallState::Running => TestState::Running,
OverallState::Failed => TestState::Failed,
OverallState::Completed => TestState::Completed,
OverallState::Confirmed => TestState::Confirmed,
OverallState::Discarded => TestState::Discarded,
},
attempt_count: status.attempt_count,
jobs: status
.jobs
.into_iter()
.map(|job| TestJob {
attempt: job.attempt,
sequence: job.sequence,
state: match job.state {
LlmJobState::Running => TestJobState::Running,
LlmJobState::Succeeded => TestJobState::Succeeded,
LlmJobState::Failed => TestJobState::Failed,
},
})
.collect(),
interim_transcript: status
.analysis
.as_ref()
.map(|analysis| analysis.envelope.analysis.transcript.clone()),
final_transcript: status.final_transcript,
labels: status
.confirmed_labels
.into_iter()
.map(|label| label.person_id)
.collect(),
errors: status.errors,
errors_truncated: status.errors_truncated,
})
}
fn retry(&self, id: TestId) -> Result<(), AdapterError> {
self.facade.retry(from_test_id(id)).map_err(adapter_error)
}
fn discard(&self, id: TestId) -> Result<(), AdapterError> {
self.facade.discard(from_test_id(id)).map_err(adapter_error)
}
fn submit_labels(&self, id: TestId, labels: Vec<String>) -> Result<(), AdapterError> {
let fragment_id = from_test_id(id);
let status = self
.facade
.status(fragment_id)
.map_err(adapter_error)?
.ok_or_else(|| adapter_error("status is unavailable".to_owned()))?;
let speakers = status
.analysis
.ok_or_else(|| adapter_error("analysis is unavailable".to_owned()))?
.envelope
.analysis
.speakers;
if speakers.len() != labels.len() {
return Err(adapter_error("label count mismatch".to_owned()));
}
let labels = speakers
.into_iter()
.zip(labels)
.map(|(speaker, person_id)| SpeakerLabelV1 {
speaker: speaker.speaker,
person_id,
})
.collect();
self.facade
.submit_labels(fragment_id, labels)
.map_err(adapter_error)
}
fn inject_error_burst(&self, id: TestId, errors: Vec<String>) -> Result<(), AdapterError> {
self.facade
.inject_error_burst(from_test_id(id), errors)
.map_err(adapter_error)
}
}
struct NoopSubsystem;
impl Subsystem for NoopSubsystem {
fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
Ok(())
}
fn reorg(&self) -> Result<(), String> {
Ok(())
}
}
fn to_test_id(id: FragmentId) -> TestId {
id.into_bytes()
}
fn from_test_id(id: TestId) -> FragmentId {
FragmentId::from_bytes(id)
}
fn adapter_error(message: String) -> AdapterError {
Box::new(std::io::Error::other(message))
}
#[test]
fn published_conformance() {
run_all(&TestFactory).expect("audio classification conformance");
}
}