115 lines
4.6 KiB
Go
115 lines
4.6 KiB
Go
package transcribe
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
|
"git.d-ma.be/mathias/tapir/internal/errkind"
|
|
"git.d-ma.be/mathias/tapir/internal/progress"
|
|
"git.d-ma.be/mathias/tapir/internal/reconcile"
|
|
)
|
|
|
|
// Ensemble transcribes the same WAV with two engines, one after the other, and has a
|
|
// local model reconcile their disagreements (ADR-034). It satisfies uploads.Transcriber,
|
|
// so the worker neither knows nor cares that two engines were involved.
|
|
//
|
|
// It degrades in steps and never turns a recoverable problem into a failed job:
|
|
// one engine down → the other's text, said so on the result; the model down → the
|
|
// base text unreconciled; both engines down → the cloud fallback, if configured, and
|
|
// recorded as such. A cancelled job stops everything and never reaches the cloud.
|
|
type Ensemble struct {
|
|
base, alt Backend
|
|
fallback Backend // nil = no cloud fallback
|
|
adj reconcile.Adjudicator
|
|
opts reconcile.Options
|
|
}
|
|
|
|
// NewEnsemble builds an Ensemble. base is the engine whose text stands where the two
|
|
// disagree and is the faster one, so it runs first; alt supplies the other reading.
|
|
func NewEnsemble(base, alt, fallback Backend, adj reconcile.Adjudicator, opts reconcile.Options) *Ensemble {
|
|
return &Ensemble{base: base, alt: alt, fallback: fallback, adj: adj, opts: opts}
|
|
}
|
|
|
|
// Plan implements uploads.Planner: the phases this ensemble reports, so the page can
|
|
// show them before they run. The fallback is a phase only when one is configured.
|
|
func (e *Ensemble) Plan() []progress.Planned {
|
|
plan := []progress.Planned{
|
|
{Key: "engine_a", Label: "Transcribing with " + e.base.Name()},
|
|
{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()},
|
|
{Key: "compare", Label: "Comparing the two transcripts"},
|
|
{Key: "decide", Label: "Deciding the differences"},
|
|
}
|
|
if e.fallback != nil {
|
|
plan = append(plan, progress.Planned{Key: "fallback", Label: "Cloud fallback (" + e.fallback.Name() + ")"})
|
|
}
|
|
return plan
|
|
}
|
|
|
|
// Transcribe implements uploads.Transcriber. Each phase is reported through the
|
|
// progress.Reporter in ctx (a no-op when nobody listens).
|
|
func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) {
|
|
rep := progress.From(ctx)
|
|
b, berr := e.run(ctx, rep, "engine_a", e.base, wavPath)
|
|
if err := ctx.Err(); err != nil {
|
|
return domain.Transcription{}, err
|
|
}
|
|
a, aerr := e.run(ctx, rep, "engine_b", e.alt, wavPath)
|
|
if err := ctx.Err(); err != nil {
|
|
return domain.Transcription{}, err
|
|
}
|
|
|
|
switch {
|
|
case berr == nil && aerr == nil:
|
|
out, err := reconcile.Reconcile(ctx, b, a, e.adj, e.opts)
|
|
if err != nil {
|
|
return domain.Transcription{}, err
|
|
}
|
|
out.Endpoint = fmt.Sprintf("%s + %s, reconciled by %s", e.base.Name(), e.alt.Name(), e.opts.Reconciler)
|
|
if rc := out.Reconciliation; rc != nil {
|
|
rep.Note("reconciled", "status", string(rc.Status), "disputes", rc.Disputes, "auto_resolved", rc.AutoResolved,
|
|
"adjudicated", rc.Adjudicated, "changed", rc.Changed, "unsure", rc.Unsure, "errors", rc.Errors)
|
|
}
|
|
return out, nil
|
|
case berr == nil:
|
|
return e.single(b, e.base.Name(), fmt.Sprintf("%s failed (%s)", e.alt.Name(), errkind.Of(aerr))), nil
|
|
case aerr == nil:
|
|
return e.single(a, e.alt.Name(), fmt.Sprintf("%s failed (%s)", e.base.Name(), errkind.Of(berr))), nil
|
|
}
|
|
|
|
both := errors.Join(fmt.Errorf("%s: %w", e.base.Name(), berr), fmt.Errorf("%s: %w", e.alt.Name(), aerr))
|
|
if e.fallback == nil {
|
|
return domain.Transcription{}, both
|
|
}
|
|
f, ferr := e.run(ctx, rep, "fallback", e.fallback, wavPath)
|
|
if ferr != nil {
|
|
return domain.Transcription{}, errors.Join(both, fmt.Errorf("%s: %w", e.fallback.Name(), ferr))
|
|
}
|
|
return e.single(f, e.fallback.Name(), "both local engines failed"), nil
|
|
}
|
|
|
|
// run is one engine call, reported as a phase. A job that was stopped is not a broken
|
|
// engine, so a cancelled context is never reported as a failed phase.
|
|
func (e *Ensemble) run(ctx context.Context, rep progress.Reporter, key string, be Backend, wav string) (domain.Transcription, error) {
|
|
rep.Begin(key, 0)
|
|
t, err := be.Transcribe(ctx, wav)
|
|
switch {
|
|
case err == nil:
|
|
rep.Done(key)
|
|
case ctx.Err() == nil:
|
|
rep.Fail(key, err)
|
|
}
|
|
return t, err
|
|
}
|
|
|
|
// single is a transcript from one engine only: nothing to reconcile, and the result
|
|
// says why, because a reader should know the second opinion was missing.
|
|
func (e *Ensemble) single(t domain.Transcription, engine, reason string) domain.Transcription {
|
|
t.Endpoint = engine
|
|
t.Reconciliation = &domain.Reconciliation{
|
|
Status: domain.ReconcileSingle, Reason: reason, Base: e.opts.BaseEngine, Alt: e.opts.AltEngine, Reconciler: e.opts.Reconciler,
|
|
}
|
|
return t
|
|
}
|