From 46bfab33432b8210882c1790d181cd10b95ea74e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 16:17:29 +0000 Subject: [PATCH] =?UTF-8?q?Lot=209=20:=20dict=C3=A9e=20vocale=20tenue=20pa?= =?UTF-8?q?r=20la=20fa=C3=A7ade=20(LOKI=5FDICTATE=3Drust)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit La façade sert /api/transcribe et /api/dictate/* quand LOKI_DICTATE=rust (expérimental ; le Go reste le défaut) et supervise elle-même le serveur de transcription sherpa-onnx : lancé à la demande sur la boucle locale, GPU masqués et sans le jeton interne, prêt dès que sa socket répond, mort au démarrage expliquée, éteint après dix minutes d'inactivité ou quand les réglages changent. Modèles Parakeet téléchargés et décompressés au fil de l'eau (tar.bz2, entrées hors du dossier refusées, liens ignorés), protocole WebSocket identique (en-tête de 8 octets, trames de 10 Kio). Côté Go, facade_dictate.go remet ces routes à la façade quand elles arrivent par le tunnel du relais, et asrShutdownVite (libération de la VRAM, optimiseur) lui demande l'arrêt par l'API interne. Parité : parity_dictate_test.go tire du Go la lecture de treize WAV, les chemins d'archive, réglages, catalogue, arguments et treize appels de route, rejoués par tests/dictate.rs ; supervision testée contre un faux sherpa-onnx qui parle le protocole. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv --- README.md | 1 + ...10-15-rust-migration-lot9-dictee-design.md | 96 ++ internal/loki/dictate_server.go | 4 + internal/loki/facade_chat.go | 5 +- internal/loki/facade_chat_test.go | 33 + internal/loki/facade_dictate.go | 52 + internal/loki/parity_dictate_test.go | 266 +++++ internal/loki/web_server.go | 3 +- rust/Cargo.lock | 37 + rust/loki-rs/Cargo.toml | 4 +- rust/loki-rs/src/api.rs | 2 + rust/loki-rs/src/dictate.rs | 1050 +++++++++++++++++ rust/loki-rs/src/lib.rs | 1 + rust/loki-rs/src/main.rs | 21 +- rust/loki-rs/src/paths.rs | 21 + rust/loki-rs/src/proxy.rs | 25 +- rust/loki-rs/tests/common/fakego.rs | 53 +- rust/loki-rs/tests/dictate.rs | 705 +++++++++++ rust/loki-rs/tests/lot2.rs | 1 + .../loki-rs/tests/parity_dictate/dictate.json | 337 ++++++ rust/loki-rs/tests/proxy.rs | 1 + 21 files changed, 2699 insertions(+), 19 deletions(-) create mode 100644 docs/superpowers/specs/2026-10-15-rust-migration-lot9-dictee-design.md create mode 100644 internal/loki/facade_dictate.go create mode 100644 internal/loki/parity_dictate_test.go create mode 100644 rust/loki-rs/src/dictate.rs create mode 100644 rust/loki-rs/tests/dictate.rs create mode 100644 rust/loki-rs/tests/parity_dictate/dictate.json diff --git a/README.md b/README.md index adfa692..3954cc0 100644 --- a/README.md +++ b/README.md @@ -229,6 +229,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur : | `LOKI_GO_PORT` | port local (127.0.0.1) du serveur Go derrière la façade | `8091` | | `LOKI_CHAT` | `rust` : la conversation est tenue par la façade Rust (expérimental, migration en cours), outils fichiers et shell compris. Le Go garde les autres outils, les pièces jointes, les fichiers et l'export. Avec `PROJ_SNAPSHOT`, `KEEP_TURN_IMAGES`, `NUDGE_IN_TOOL`, `SIDE_SLOT`, `SLOT_PERSIST`, `PREWARM` ou des motifs `retry_patterns`, le Go garde la conversation (décidé au démarrage) | Go | | `LOKI_MODELS` | `rust` : les routes des modèles — liste, dossiers, suppression, téléchargements, catalogue Hugging Face — sont servies par la façade Rust (expérimental, migration en cours) | Go | +| `LOKI_DICTATE` | `rust` : la dictée vocale (`/api/transcribe`, `/api/dictate/*`) et son serveur de transcription sont tenus par la façade Rust (expérimental, migration en cours) | Go | **Accès par nom de domaine.** Sans clé de pilotage, l'API n'accepte que les hôtes locaux (IP, `localhost`, nom sans point, `.local`/`.lan`…) et refuse diff --git a/docs/superpowers/specs/2026-10-15-rust-migration-lot9-dictee-design.md b/docs/superpowers/specs/2026-10-15-rust-migration-lot9-dictee-design.md new file mode 100644 index 0000000..167450c --- /dev/null +++ b/docs/superpowers/specs/2026-10-15-rust-migration-lot9-dictee-design.md @@ -0,0 +1,96 @@ +# Lot 9 — dictée vocale tenue par la façade + +## Contexte + +La dictée vocale vivait dans le Go : + +- `dictate_config.go` : réglages ; +- `dictate_models.go` : catalogue Parakeet, téléchargement et extraction ; +- `dictate_server.go` : supervision de `sherpa-onnx-offline-websocket-server` + et protocole WebSocket ; +- `web_transcribe.go` : routes. + +Lot 9 : la façade la tient quand `LOKI_DICTATE=rust` (expérimental ; sans la +variable rien ne change). + +## Ce que sert la façade (`rust/loki-rs/src/dictate.rs`) + +| Route | Rôle | +|---|---| +| `POST /api/transcribe` | un WAV (≥ 1000 octets, ≤ 32 Mio) entre, le texte sort ; modèle absent : 503 `{downloading, pct}` et téléchargement lancé (relancé après un échec, qui est dit) | +| `GET\|POST /api/dictate/config` | modèle (validé contre le catalogue) et réactivité ; enregistrer arrête le serveur, qui repart au prochain mot avec le nouveau modèle | +| `GET /api/dictate/models` | catalogue trié par taille, présence des quatre fichiers | +| `GET\|POST /api/dictate/models/download` | un téléchargement à la fois, progression sur les octets reçus | +| `GET /api/dictate/state` | serveur allumé (et depuis quand), modèle, présence, téléchargement | +| `POST /internal/facade/v1/dictate/stop` | (le Go, jeton interne) arrêt sans attendre derrière un démarrage | + +Même mécanique que le Go : + +**Archive `.tar.bz2`** +- Décompressée au fil du téléchargement, dans un dossier `.part` renommé à la fin. +- Entrées absolues ou qui remontent refusées, liens ignorés. + +**Serveur de transcription** +- Lancé à la demande sur un port libre de la boucle locale, GPU masqués, sans + le jeton interne dans son environnement. +- Prêt dès que sa socket répond. +- Une mort au démarrage est rapportée avec sa cause et la dernière ligne du + journal ; un processus qui ne répond pas fait échouer le démarrage au bout + de 2 minutes. +- Éteint après dix minutes d'inactivité. +- `shutdown_fast` tue le processus qui démarre au lieu d'attendre le verrou. + +**Protocole** +- En-tête de 8 octets (fréquence, nombre d'octets), échantillons float32 en + trames de 10 Kio, une trame texte en retour. +- Annotations (`[musique]`, `(rires)`) écartées. + +## Côté Go (`facade_dictate.go`) + +- `withFacadeChat` remet les routes du tunnel et de la redistribution E2E à la + façade. Si elle ne répond pas : 502 « dictée indisponible : … ». +- `asrShutdownVite` passe par la façade quand elle tient la dictée : son + serveur n'est pas un enfant de ce process. Cela couvre la libération de la + VRAM et l'arrêt du moteur principal par l'optimiseur. + +## Écarts assumés + +- Messages d'erreur de transport (WebSocket, HTTP) et de décompression au + format des crates Rust. +- Fin de connexion sans réponse : « lecture du résultat : connexion fermée ». + +## Parité + +`internal/loki/parity_dictate_test.go` tire du Go ce journal : + +- treize WAV : mono, bloc `LIST` intercalé, stéréo moyenné, flottant 32 bits, + données tronquées, impaires, vides, et les refus ; +- les chemins d'archive ; +- les marqueurs ; +- les réglages par défaut ; +- les arguments du serveur ; +- les adresses des modèles ; +- la présence ; +- treize appels de route. + +Le tout est rangé dans `tests/parity_dictate/dictate.json` et rejoué par +`tests/dictate.rs`, les routes par la façade. Une mutation (premier canal au +lieu de la moyenne) fait échouer le rejeu. + +`tests/dictate.rs` couvre aussi : + +- le protocole contre un faux serveur WebSocket (en-tête, découpage + 10240 + 10240 + 3528, marqueurs, réponse illisible, serveur injoignable) ; +- l'extraction (lien ignoré, entrée qui remonte refusée sans rien laisser, + archive corrompue) ; +- le téléchargement depuis un faux serveur (100 %, HTTP 404, modèle inconnu, + un seul à la fois) ; +- la supervision d'un faux sherpa-onnx (script Python qui parle le + protocole) : + - transcription de bout en bout par la façade ; + - même serveur réutilisé ; + - arrêt demandé par le Go, et refusé sans le jeton ; + - réglage changé ; +- les démarrages ratés : sortie en code 3 avec sa dernière ligne, serveur + muet, arrêt « vite » qui tue le démarrage, binaire introuvable, modèle + absent. diff --git a/internal/loki/dictate_server.go b/internal/loki/dictate_server.go index 18a209d..9f727de 100644 --- a/internal/loki/dictate_server.go +++ b/internal/loki/dictate_server.go @@ -414,6 +414,10 @@ func asrShutdown() { // normale. La dictée qui attendait reçoit une erreur de démarrage — c'est // l'utilisateur qui vient de demander la place, pas une panne. func asrShutdownVite() { + // Dictée tenue par la façade (LOKI_DICTATE=rust) : son serveur est le sien. + if facadeDictateStop() { + return + } for i := 0; i < 20; i++ { if wsrvMu.TryLock() { asrStopLocked() diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go index 453d58f..a264a15 100644 --- a/internal/loki/facade_chat.go +++ b/internal/loki/facade_chat.go @@ -141,7 +141,8 @@ var facadeChatRoutes = map[string]bool{ } // withFacadeChat remet la requête à la façade quand elle tient la -// conversation — ou les routes des modèles (facade_models.go). +// conversation — ou les routes des modèles (facade_models.go), ou la dictée +// (facade_dictate.go). // Posé APRÈS requireWebAuth : seule une requête authentifiée ici part, avec le // jeton interne qui en tient lieu là-bas. func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { @@ -150,6 +151,8 @@ func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { case facadeChatRoutes[path]: case facadeModelRoutes[path]: held, unavailable = facadeModels, "modèles indisponibles : " + case facadeDictateRoutes[path]: + held, unavailable = facadeDictate, "dictée indisponible : " default: return h } diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go index 048dcfb..ef1e3d6 100644 --- a/internal/loki/facade_chat_test.go +++ b/internal/loki/facade_chat_test.go @@ -379,3 +379,36 @@ func TestFacadeTelechargementsGardeLesPart(t *testing.T) { t.Fatalf(".part orphelin gardé : %v", err) } } + +// LOKI_DICTATE=rust : les routes de la dictée partent à la façade, et arrêter +// la dictée (libération de la VRAM, optimiseur) le lui demande au lieu de +// toucher au serveur de ce process. Sans la variable, rien ne change. +func TestFacadeDictee(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + t.Setenv(facadeChatEnv, "") + t.Setenv(facadeDictateEnv, "rust") + local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") } + for p := range facadeDictateRoutes { + w := httptest.NewRecorder() + withFacadeChat(p, local)(w, httptest.NewRequest("POST", p, strings.NewReader("{}"))) + if w.Body.String() != f.reply { + t.Fatalf("%s : réponse %q", p, w.Body.String()) + } + } + asrShutdownVite() + got := f.recu() + if len(got) != len(facadeDictateRoutes)+1 || got[len(got)-1] != "POST /internal/facade/v1/dictate/stop jeton {}" { + t.Fatalf("reçu par la façade : %v", got) + } + t.Setenv(facadeDictateEnv, "") + w := httptest.NewRecorder() + withFacadeChat("/api/transcribe", local)(w, httptest.NewRequest("POST", "/api/transcribe", nil)) + if w.Body.String() != "go" { + t.Fatalf("relayé sans la variable : %q", w.Body.String()) + } + asrShutdownVite() + if n := len(f.recu()); n != len(facadeDictateRoutes)+1 { + t.Fatalf("arrêt demandé à la façade sans la variable (%d appels)", n) + } +} diff --git a/internal/loki/facade_dictate.go b/internal/loki/facade_dictate.go new file mode 100644 index 0000000..a0f8afa --- /dev/null +++ b/internal/loki/facade_dictate.go @@ -0,0 +1,52 @@ +package loki + +import ( + "fmt" + "os" + "strings" +) + +// facade_dictate.go — dictée vocale tenue par la façade Rust (LOKI_DICTATE=rust, +// lot 9 de la migration ; expérimental, le Go reste le défaut). +// +// La façade sert elle-même /api/transcribe et /api/dictate/* sur le port +// public, et supervise le serveur de transcription. Deux choses restent à +// régler ici : +// +// - le tunnel du relais et sa redistribution E2E arrivent dans ce process : +// ces routes sont remises à la façade (withFacadeChat) ; +// - libérer la VRAM (web_vram.go) et l'optimiseur (backend_tune_run.go) +// arrêtent la dictée : c'est alors à la façade qu'on le demande, son +// serveur n'étant pas un enfant de ce process. + +const facadeDictateEnv = "LOKI_DICTATE" + +// facadeDictate : la façade tient-elle la dictée ? Il faut aussi son adresse +// et le jeton interne, sans lesquels rien ne peut lui être remis. +func facadeDictate() bool { + return strings.EqualFold(strings.TrimSpace(os.Getenv(facadeDictateEnv)), "rust") && + facadeAddr() != "" && internalToken() != "" +} + +// facadeDictateRoutes : les routes que la façade sert alors (dictate.rs). +var facadeDictateRoutes = map[string]bool{ + "/api/transcribe": true, + "/api/dictate/config": true, + "/api/dictate/models": true, + "/api/dictate/models/download": true, + "/api/dictate/state": true, +} + +// facadeDictateStop demande à la façade d'arrêter son serveur de dictée, sans +// attendre derrière un démarrage (elle tue le processus qui démarre). Rend +// false quand la façade ne tient pas la dictée : l'appelant arrête alors le +// serveur de ce process. +func facadeDictateStop() bool { + if !facadeDictate() { + return false + } + if err := facadePost("/internal/facade/v1/dictate/stop", map[string]any{}); err != nil { + fmt.Printf("[dictée] arrêt demandé à la façade : %v\n", err) + } + return true +} diff --git a/internal/loki/parity_dictate_test.go b/internal/loki/parity_dictate_test.go new file mode 100644 index 0000000..1649d06 --- /dev/null +++ b/internal/loki/parity_dictate_test.go @@ -0,0 +1,266 @@ +package loki + +import ( + "encoding/binary" + "encoding/json" + "math" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "testing" +) + +// parity_dictate_test.go — journal de référence de la dictée vocale (lot 9 de +// la migration Rust) : lecture des WAV, chemins d'archive, marqueurs, +// réglages, catalogue, arguments du serveur, et les routes qui ne lancent +// rien (réglages, catalogue, état, refus). Rangé dans +// rust/loki-rs/tests/parity_dictate/dictate.json ; tests/dictate.rs le rejoue. +// LOKI_HOME réécrit en {{HOME}}. +// +// LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteDictee + +const parityDictateFile = "../../rust/loki-rs/tests/parity_dictate/dictate.json" + +type parityWav struct { + Name string `json:"name"` + // WAV en octets (JSON : base64). + In []byte `json:"in"` + Rate int `json:"rate,omitempty"` + Ech []float32 `json:"ech,omitempty"` + Err string `json:"err,omitempty"` +} + +type parityDictateCall struct { + Method string `json:"method"` + Path string `json:"path"` + Body string `json:"body,omitempty"` + Status int `json:"status"` + Out any `json:"out"` +} + +type parityDictateFixture struct { + Wav []parityWav `json:"wav"` + SafeRel []struct { + In string `json:"in"` + Out string `json:"out,omitempty"` + Err string `json:"err,omitempty"` + } `json:"safe_rel"` + Marker map[string]bool `json:"marker"` + Defaults []map[string]string `json:"defaults"` + LastLine map[string]string `json:"last_line"` + Args []string `json:"args"` + ArgsErr string `json:"args_err"` + URLs map[string]string `json:"urls"` + Calls []parityDictateCall `json:"calls"` + Present map[string]bool `json:"present"` +} + +// parityWavOf : un WAV avec les blocs donnés après l'en-tête RIFF/WAVE. +func parityWavOf(chunks ...[]byte) []byte { + b := []byte("RIFF\x00\x00\x00\x00WAVE") + for _, c := range chunks { + b = append(b, c...) + } + binary.LittleEndian.PutUint32(b[4:8], uint32(len(b)-8)) + return b +} + +func parityChunk(id string, body []byte) []byte { + h := make([]byte, 8) + copy(h, id) + binary.LittleEndian.PutUint32(h[4:], uint32(len(body))) + out := append(h, body...) + if len(body)%2 == 1 { + out = append(out, 0) + } + return out +} + +func parityFmt(format, canaux, taux, bits int) []byte { + b := make([]byte, 16) + binary.LittleEndian.PutUint16(b[0:], uint16(format)) + binary.LittleEndian.PutUint16(b[2:], uint16(canaux)) + binary.LittleEndian.PutUint32(b[4:], uint32(taux)) + binary.LittleEndian.PutUint32(b[8:], uint32(taux*canaux*bits/8)) + binary.LittleEndian.PutUint16(b[12:], uint16(canaux*bits/8)) + binary.LittleEndian.PutUint16(b[14:], uint16(bits)) + return parityChunk("fmt ", b) +} + +func parityPCM16(v ...int16) []byte { + b := make([]byte, 2*len(v)) + for i, x := range v { + binary.LittleEndian.PutUint16(b[2*i:], uint16(x)) + } + return b +} + +func TestPariteDictee(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("journal tiré sous Unix") + } + home := testHome(t) + if r, err := filepath.EvalSymlinks(home); err == nil { + home = r + } + t.Setenv("LOKI_HOME", home) + norm := func(s string) string { return strings.ReplaceAll(s, home, "{{HOME}}") } + var fx parityDictateFixture + + f32 := make([]byte, 12) + for i, v := range []float32{0.25, -1, 0.5} { + binary.LittleEndian.PutUint32(f32[4*i:], math.Float32bits(v)) + } + cases := []struct { + name string + in []byte + }{ + {"mono 16 bits", parityWavOf(parityFmt(1, 1, 16000, 16), parityChunk("data", parityPCM16(0, 16384, -16384, 32767, -32768)))}, + {"bloc LIST intercalé", parityWavOf(parityChunk("LIST", []byte("INFOxyz")), parityFmt(1, 1, 16000, 16), parityChunk("data", parityPCM16(100, -100)))}, + {"stéréo moyenné", parityWavOf(parityFmt(1, 2, 48000, 16), parityChunk("data", parityPCM16(0, 32767, -32768, 32767)))}, + {"flottant 32 bits", parityWavOf(parityFmt(3, 1, 16000, 32), parityChunk("data", f32))}, + {"data tronqué", parityWavOf(parityFmt(1, 1, 16000, 16), parityChunk("data", parityPCM16(1, 2, 3)))[:12+24+8+4]}, + {"data impair", parityWavOf(parityFmt(1, 1, 8000, 16), parityChunk("data", []byte{1, 2, 3}))}, + {"data vide", parityWavOf(parityFmt(1, 1, 16000, 16), parityChunk("data", nil))}, + {"24 bits", parityWavOf(parityFmt(1, 1, 16000, 24), parityChunk("data", make([]byte, 6)))}, + {"data avant fmt", parityWavOf(parityChunk("data", parityPCM16(1)), parityFmt(1, 1, 16000, 16))}, + {"fmt tronqué", parityWavOf(parityChunk("fmt ", make([]byte, 8)))}, + {"sans data", parityWavOf(parityFmt(1, 1, 16000, 16))}, + {"pas un WAV", []byte("ceci n'est pas un fichier audio du tout")}, + {"vide", []byte{}}, + } + for _, c := range cases { + w := parityWav{Name: c.name, In: c.in} + taux, ech, err := wavVersFloat32(c.in) + if err != nil { + w.Err = err.Error() + } else { + w.Rate, w.Ech = taux, ech + if w.Ech == nil { + w.Ech = []float32{} + } + } + fx.Wav = append(fx.Wav, w) + } + + for _, in := range []string{"", "racine", "racine/", "racine/encoder.onnx", "racine/sous/tokens.txt", + "/abs/x", "racine/../x", `racine\a\b.txt`, " racine/a ", "..", "a/b/../c"} { + out, err := asrSafeRel(in) + c := struct { + In string `json:"in"` + Out string `json:"out,omitempty"` + Err string `json:"err,omitempty"` + }{In: in, Out: out} + if err != nil { + c.Err = err.Error() + } + fx.SafeRel = append(fx.SafeRel, c) + } + fx.Marker = map[string]bool{} + for _, s := range []string{"", " ", "[musique]", "(silence)", " (rires) ", "bonjour", "[a] b", "(x"} { + fx.Marker[s] = asrMarqueur(s) + } + for _, c := range []DictateCfg{{}, {Model: "ggml-base.bin"}, {Model: "parakeet-tdt-0.6b-v2", Reactivity: "court"}} { + d := c.withDefauts() + fx.Defaults = append(fx.Defaults, map[string]string{"model": c.Model, "reactivity": c.Reactivity, + "out_model": d.Model, "out_reactivity": d.Reactivity}) + } + fx.LastLine = map[string]string{} + for _, s := range []string{"", "a\nb\n\n ", "un seul", " \n x \n"} { + fx.LastLine[s] = lastLine(s) + } + args, _ := asrServerArgs(DictateCfg{Model: "parakeet-tdt-0.6b-v3"}, 4567) + for _, a := range args { + fx.Args = append(fx.Args, norm(a)) + } + if _, err := asrServerArgs(DictateCfg{Model: "x"}, 1); err != nil { + fx.ArgsErr = err.Error() + } + fx.URLs = map[string]string{} + for id := range asrCatalogue { + fx.URLs[id] = asrModelURLFor(id) + } + fx.URLs["inconnu"] = asrModelURLFor("inconnu") + + // v2 installé (quatre fichiers non vides), v3 à moitié. + for _, f := range asrFichiers { + p := asrModelFile("parakeet-tdt-0.6b-v2", f) + _ = os.MkdirAll(filepath.Dir(p), 0o755) + _ = os.WriteFile(p, []byte("x"), 0o644) + } + p := asrModelFile("parakeet-tdt-0.6b-v3", "encoder.int8.onnx") + _ = os.MkdirAll(filepath.Dir(p), 0o755) + _ = os.WriteFile(p, []byte("x"), 0o644) + _ = os.WriteFile(asrModelFile("parakeet-tdt-0.6b-v3", "tokens.txt"), nil, 0o644) + fx.Present = map[string]bool{} + for id := range asrCatalogue { + fx.Present[id] = asrModelPresent(id) + } + fx.Present["inconnu"] = asrModelPresent("inconnu") + + call := func(method, path, body string) parityDictateCall { + w := httptest.NewRecorder() + r := httptest.NewRequest(method, path, strings.NewReader(body)) + switch path { + case "/api/dictate/config": + handleDictateConfig(w, r) + case "/api/dictate/models": + handleDictateModels(w, r) + case "/api/dictate/models/download": + handleDictateDownload(w, r) + case "/api/dictate/state": + handleDictateState(w, r) + case "/api/transcribe": + handleTranscribe(w, r) + default: + t.Fatalf("route inattendue %s", path) + } + var out any + if err := json.Unmarshal([]byte(norm(w.Body.String())), &out); err != nil { + t.Fatalf("%s %s : %v (%s)", method, path, err, w.Body.String()) + } + return parityDictateCall{Method: method, Path: path, Body: body, Status: w.Code, Out: out} + } + for _, c := range []struct{ m, p, b string }{ + {"GET", "/api/dictate/config", ""}, + {"POST", "/api/dictate/config", `pas du json`}, + {"POST", "/api/dictate/config", `{"model":"ggml-base.bin"}`}, + {"POST", "/api/dictate/config", `{"model":"parakeet-tdt-0.6b-v2"}`}, + {"GET", "/api/dictate/config", ""}, + {"POST", "/api/dictate/config", `{"model":"parakeet-tdt-0.6b-v3","reactivity":"long","autre":1}`}, + {"GET", "/api/dictate/config", ""}, + {"GET", "/api/dictate/models", ""}, + {"GET", "/api/dictate/models/download", ""}, + {"POST", "/api/dictate/models/download", `{"id":" inconnu "}`}, + {"POST", "/api/dictate/models/download", `x`}, + {"GET", "/api/dictate/state", ""}, + {"POST", "/api/transcribe", `trop court`}, + } { + fx.Calls = append(fx.Calls, call(c.m, c.p, c.b)) + } + + got, err := parityJSON(fx) + if err != nil { + t.Fatal(err) + } + if os.Getenv("LOKI_PARITY_UPDATE") == "1" { + if err := os.MkdirAll(filepath.Dir(parityDictateFile), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(parityDictateFile, got, 0o644); err != nil { + t.Fatal(err) + } + return + } + want, err := os.ReadFile(parityDictateFile) + if err != nil { + t.Fatalf("journal de référence absent (%v) : LOKI_PARITY_UPDATE=1 pour le créer", err) + } + if string(got) != string(want) { + t.Errorf("la dictée ne produit plus le journal rangé dans %s.\n"+ + "Si c'est voulu : LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteDictee,\n"+ + "puis cargo test --test dictate dit si le port Rust suit.\n--- obtenu ---\n%s", parityDictateFile, got) + } +} diff --git a/internal/loki/web_server.go b/internal/loki/web_server.go index a1bd310..9b35c1b 100644 --- a/internal/loki/web_server.go +++ b/internal/loki/web_server.go @@ -220,7 +220,8 @@ func newWebMux() *http.ServeMux { // api enregistre une route /api/* protégée par la clé de pilotage (web_auth.go). // withFacadeChat : routes du chat remises à la façade quand elle tient la // conversation (LOKI_CHAT=rust, facade_chat.go), celles des modèles quand - // elle les tient (LOKI_MODELS=rust, facade_models.go) ; sans effet sinon. + // elle les tient (LOKI_MODELS=rust, facade_models.go), celles de la dictée + // (LOKI_DICTATE=rust, facade_dictate.go) ; sans effet sinon. api := func(path string, h http.HandlerFunc) { mux.HandleFunc(path, requireWebAuth(withFacadeChat(path, h))) } api("/api/ping", handlePing) api("/api/status", handleStatus) diff --git a/rust/Cargo.lock b/rust/Cargo.lock index c003d08..7221873 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -56,6 +56,15 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "bzip2" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3a53fac24f34a81bc9954b5d6cfce0c21e18ec6959f44f56e8e90e4bb7c346c" +dependencies = [ + "libbz2-rs-sys", +] + [[package]] name = "cc" version = "1.8.0" @@ -192,6 +201,16 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.14" @@ -639,6 +658,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "libbz2-rs-sys" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34b357333733e8260735ba5894eb928c02ecc69c78715f01a8019e7fa7f2db4c" + [[package]] name = "libc" version = "0.2.190" @@ -668,6 +693,7 @@ name = "loki-rs" version = "0.1.0" dependencies = [ "bytes", + "bzip2", "flate2", "futures-util", "getrandom 0.4.3", @@ -683,6 +709,7 @@ dependencies = [ "serde_json", "sha1", "sha2", + "tar", "tempfile", "tokio", "tokio-tungstenite", @@ -1198,6 +1225,16 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", +] + [[package]] name = "tempfile" version = "3.27.0" diff --git a/rust/loki-rs/Cargo.toml b/rust/loki-rs/Cargo.toml index 191251e..9d5e62e 100644 --- a/rust/loki-rs/Cargo.toml +++ b/rust/loki-rs/Cargo.toml @@ -5,6 +5,7 @@ edition = "2024" [dependencies] bytes = "1.12.1" +bzip2 = "0.6.1" flate2 = "1.1.10" futures-util = "0.3.34" getrandom = "0.4.3" @@ -20,7 +21,9 @@ serde = { version = "1.0.229", features = ["derive"] } serde_json = { version = "1.0.154", features = ["raw_value"] } sha1 = "0.11.0" sha2 = "0.11.1" +tar = { version = "0.4.46", default-features = false } tokio = { version = "1.53.2", features = ["rt-multi-thread", "macros", "net", "io-util", "time", "signal", "process", "sync", "fs"] } +tokio-tungstenite = "0.30.0" [features] # Crypto TLS (ring) pour les API externes en HTTPS. Activée par défaut ; le @@ -35,4 +38,3 @@ libc = "0.2" tokio = { version = "1.53.2", features = ["test-util"] } futures-util = "0.3.34" tempfile = "3.27.0" -tokio-tungstenite = "0.30.0" diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs index 870e36c..6f5cde0 100644 --- a/rust/loki-rs/src/api.rs +++ b/rust/loki-rs/src/api.rs @@ -30,6 +30,8 @@ pub struct Ctx { /// catalogue Hugging Face) tenues par la façade (`LOKI_MODELS=rust`) ; /// `None` : au Go. pub models: Option>, + /// Dictée vocale tenue par la façade (`LOKI_DICTATE=rust`) ; `None` : au Go. + pub dictate: Option>, /// Jeton de l'API interne : une requête qui le porte vient du Go. pub internal_token: String, } diff --git a/rust/loki-rs/src/dictate.rs b/rust/loki-rs/src/dictate.rs new file mode 100644 index 0000000..962b0e3 --- /dev/null +++ b/rust/loki-rs/src/dictate.rs @@ -0,0 +1,1050 @@ +//! Dictée vocale — port de `dictate_config.go`, `dictate_models.go`, +//! `dictate_server.go` et `web_transcribe.go` (lot 9 de la migration), servie +//! par la façade quand `LOKI_DICTATE=rust` (expérimental ; le Go par défaut) : +//! +//! - `POST /api/transcribe` : un WAV entre, le texte sort ; +//! - `GET|POST /api/dictate/config` : modèle et réactivité ; +//! - `GET /api/dictate/models` : catalogue, présence sur le disque ; +//! - `GET|POST /api/dictate/models/download` : téléchargement d'un modèle ; +//! - `GET /api/dictate/state` : serveur allumé, modèle, téléchargement. +//! +//! La transcription est locale : Parakeet servi par +//! `sherpa-onnx-offline-websocket-server`, lancé à la demande, arrêté après +//! dix minutes d'inactivité ou quand les réglages changent. Le Go demande son +//! arrêt (libération de la VRAM, optimiseur) par +//! `POST /internal/facade/v1/dictate/stop`. +//! +//! Écarts assumés : messages d'erreur de transport (WebSocket, HTTP) et de +//! décompression au format des crates Rust ; corps JSON illisible : « requête +//! illisible » comme le Go, quelle que soit la faute. + +use std::io::Read; +use std::path::{Path, PathBuf}; +use std::process::{ExitStatus, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use futures_util::{SinkExt, StreamExt}; +use http_body_util::{BodyExt, Limited}; +use hyper::body::Incoming; +use hyper::{Method, Request, Response, StatusCode}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use tokio::io::AsyncReadExt; +use tokio_tungstenite::tungstenite::Message; + +use crate::api::Ctx; +use crate::auth::json as send_json; +use crate::proxy::Body; +use crate::store::BK_STATE; + +pub const DICTATE_CFG_KEY: &str = "dictate_cfg"; +/// Les archives sherpa-onnx officielles. +pub const ASR_MODEL_BASE: &str = + "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/"; +/// Les quatre fichiers d'un modèle utilisable. +pub const ASR_FICHIERS: [&str; 4] = [ + "encoder.int8.onnx", + "decoder.int8.onnx", + "joiner.int8.onnx", + "tokens.txt", +]; +const DEFAULT_MODEL: &str = "parakeet-tdt-0.6b-v3"; +/// En-tête du protocole : fréquence, puis nombre d'octets d'audio. +const ENTETE: usize = 8; +/// Taille d'une trame WebSocket envoyée (celle du client de référence). +const TRAME_MAX: usize = 10240; +/// Le serveur s'éteint après ce délai sans dictée. +pub const ASR_IDLE: Duration = Duration::from_secs(600); +/// Plafond d'attente au démarrage (modèle de 600 Mo sur un disque lent). +pub const ASR_READY: Duration = Duration::from_secs(120); + +/// Routes servies ici. +pub fn handles(path: &str) -> bool { + matches!( + path, + "/api/transcribe" + | "/api/dictate/config" + | "/api/dictate/models" + | "/api/dictate/models/download" + | "/api/dictate/state" + ) +} + +/// La route interne par laquelle le Go demande l'arrêt du serveur. +pub const INTERNAL_STOP: &str = "/internal/facade/v1/dictate/stop"; + +// --------------------------------------------------------------------------- +// Catalogue et réglages +// --------------------------------------------------------------------------- + +/// Un modèle du catalogue (`asrModel`). +pub struct AsrModel { + pub id: &'static str, + pub nom: &'static str, + pub dossier: &'static str, + pub octets: i64, + pub langues: &'static str, +} + +pub const CATALOGUE: [AsrModel; 2] = [ + AsrModel { + id: "parakeet-tdt-0.6b-v3", + nom: "Parakeet v3 — recommandé", + dossier: "sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8", + octets: 487_170_055, + langues: "25 langues européennes, français inclus — détection automatique", + }, + AsrModel { + id: "parakeet-tdt-0.6b-v2", + nom: "Parakeet v2 — anglais seul", + dossier: "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8", + octets: 482_468_385, + langues: "anglais uniquement", + }, +]; + +fn model(id: &str) -> Option<&'static AsrModel> { + CATALOGUE.iter().find(|m| m.id == id) +} + +/// Réglages (`DictateCfg`), rangés dans `state/dictate_cfg`. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct DictateCfg { + #[serde(default)] + pub model: String, + #[serde(default)] + pub reactivity: String, +} + +impl DictateCfg { + /// `withDefauts` : modèle hors catalogue → v3 ; réactivité vide → moyen. + pub fn with_defauts(mut self) -> Self { + if model(&self.model).is_none() { + self.model = DEFAULT_MODEL.into(); + } + if self.reactivity.is_empty() { + self.reactivity = "moyen".into(); + } + self + } +} + +/// `asrRoot` : `LOKI_HOME/asr`. +pub fn asr_root(home: &Path) -> PathBuf { + home.join("asr") +} + +/// `asrModelDirFor` : `None` si l'identifiant est inconnu. +pub fn model_dir_for(home: &Path, id: &str) -> Option { + model(id).map(|m| asr_root(home).join(m.dossier)) +} + +/// `asrModelURLFor` : "" si l'identifiant est inconnu. +pub fn model_url_for(base: &str, id: &str) -> String { + model(id).map_or(String::new(), |m| format!("{base}{}.tar.bz2", m.dossier)) +} + +/// `asrModelPresent` : les quatre fichiers, non vides. +pub fn model_present(home: &Path, id: &str) -> bool { + let Some(dir) = model_dir_for(home, id) else { + return false; + }; + ASR_FICHIERS + .iter() + .all(|f| std::fs::metadata(dir.join(f)).is_ok_and(|m| m.len() > 0)) +} + +/// `asrCatalogueTrie` : du plus léger au plus lourd. +pub fn catalogue_trie(home: &Path) -> Value { + let mut v: Vec<&AsrModel> = CATALOGUE.iter().collect(); + v.sort_by_key(|m| m.octets); + Value::Array( + v.into_iter() + .map(|m| { + json!({ "id": m.id, "nom": m.nom, "octets": m.octets, "langues": m.langues, "present": model_present(home, m.id) }) + }) + .collect(), + ) +} + +/// `asrServerArgs` : `--clef=valeur`, écoute sur la boucle locale seulement. +pub fn server_args(home: &Path, cfg: &DictateCfg, port: u16) -> Result, String> { + let dir = model_dir_for(home, &cfg.model) + .ok_or_else(|| format!("modèle de dictée inconnu : {}", cfg.model))?; + let f = |n: &str| dir.join(n).to_string_lossy().into_owned(); + Ok(vec![ + format!("--encoder={}", f("encoder.int8.onnx")), + format!("--decoder={}", f("decoder.int8.onnx")), + format!("--joiner={}", f("joiner.int8.onnx")), + format!("--tokens={}", f("tokens.txt")), + "--model-type=nemo_transducer".into(), + format!("--port={port}"), + "--host=127.0.0.1".into(), + ]) +} + +/// `asrMarqueur` : vide, ou annotation entre crochets ou parenthèses. +pub fn marqueur(s: &str) -> bool { + let s = s.trim(); + s.is_empty() + || (s.starts_with('[') && s.ends_with(']')) + || (s.starts_with('(') && s.ends_with(')')) +} + +/// `lastLine` : la dernière ligne non vide. +pub fn last_line(s: &str) -> String { + s.trim() + .split('\n') + .rev() + .map(str::trim) + .find(|l| !l.is_empty()) + .unwrap_or("") + .to_string() +} + +// --------------------------------------------------------------------------- +// WAV +// --------------------------------------------------------------------------- + +fn u16le(b: &[u8], i: usize) -> u16 { + u16::from_le_bytes([b[i], b[i + 1]]) +} + +fn u32le(b: &[u8], i: usize) -> u32 { + u32::from_le_bytes([b[i], b[i + 1], b[i + 2], b[i + 3]]) +} + +/// `wavVersFloat32` : fréquence et échantillons dans [-1, 1], en parcourant +/// les blocs (`LIST` & co peuvent précéder `data`). +pub fn wav_to_f32(wav: &[u8]) -> Result<(u32, Vec), String> { + if wav.len() < 12 || &wav[0..4] != b"RIFF" || &wav[8..12] != b"WAVE" { + return Err("audio illisible : ce n'est pas un WAV".into()); + } + let (mut canaux, mut bits, mut format, mut taux) = (0usize, 0u16, 0u16, 0u32); + let mut pos = 12usize; + while pos + 8 <= wav.len() { + let id = &wav[pos..pos + 4]; + let mut taille = u32le(wav, pos + 4) as usize; + let corps = pos + 8; + if corps + taille > wav.len() { + // Bloc plus grand que le fichier : ce qui reste. + taille = wav.len() - corps; + } + match id { + b"fmt " => { + if taille < 16 { + return Err("audio illisible : bloc fmt tronqué".into()); + } + format = u16le(wav, corps); + canaux = u16le(wav, corps + 2) as usize; + taux = u32le(wav, corps + 4); + bits = u16le(wav, corps + 14); + } + b"data" => { + if canaux == 0 || taux == 0 { + return Err("audio illisible : bloc data avant fmt".into()); + } + let ech = pcm_to_f32(&wav[corps..corps + taille], format, bits, canaux)?; + return Ok((taux, ech)); + } + _ => {} + } + // Blocs alignés sur un nombre pair d'octets. + pos = corps + taille + taille % 2; + } + Err("audio illisible : aucun bloc data".into()) +} + +/// `pcmVersFloat32` : canaux moyennés (calcul en f64 comme le Go). +pub fn pcm_to_f32(data: &[u8], format: u16, bits: u16, canaux: usize) -> Result, String> { + match (format, bits) { + (1, 16) => { + let n = data.len() / 2 / canaux; + Ok((0..n) + .map(|i| { + let somme: f64 = (0..canaux) + .map(|c| { + i16::from_le_bytes([ + data[2 * (i * canaux + c)], + data[2 * (i * canaux + c) + 1], + ]) as f64 + / 32768.0 + }) + .sum(); + (somme / canaux as f64) as f32 + }) + .collect()) + } + (3, 32) => { + let n = data.len() / 4 / canaux; + Ok((0..n) + .map(|i| { + let somme: f64 = (0..canaux) + .map(|c| f32::from_bits(u32le(data, 4 * (i * canaux + c))) as f64) + .sum(); + (somme / canaux as f64) as f32 + }) + .collect()) + } + _ => Err(format!( + "format audio non géré (format {format}, {bits} bits) — attendu PCM 16 bits" + )), + } +} + +// --------------------------------------------------------------------------- +// Protocole +// --------------------------------------------------------------------------- + +/// `asrInferSur` : une trame binaire (en-tête + échantillons f32, découpée +/// en trames de 10 Kio), une trame texte en retour. +pub async fn infer_at(adresse: &str, wav: &[u8]) -> Result { + let (taux, ech) = wav_to_f32(wav)?; + if ech.is_empty() { + return Ok(String::new()); + } + let cfg = tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default() + .max_message_size(None) + .max_frame_size(None); + let (mut ws, _) = tokio_tungstenite::connect_async_with_config(adresse, Some(cfg), false) + .await + .map_err(|e| format!("connexion au serveur de dictée : {e}"))?; + let mut corps = Vec::with_capacity(ENTETE + 4 * ech.len()); + corps.extend_from_slice(&taux.to_le_bytes()); + corps.extend_from_slice(&((4 * ech.len()) as u32).to_le_bytes()); + for v in &ech { + corps.extend_from_slice(&v.to_bits().to_le_bytes()); + } + for trame in corps.chunks(TRAME_MAX) { + ws.send(Message::Binary(Bytes::copy_from_slice(trame))) + .await + .map_err(|e| format!("envoi de l'audio : {e}"))?; + } + let brut = loop { + match ws.next().await { + Some(Ok(Message::Text(t))) => break t.as_bytes().to_vec(), + Some(Ok(Message::Binary(b))) => break b.to_vec(), + Some(Ok(Message::Close(_))) | None => { + return Err("lecture du résultat : connexion fermée".into()); + } + Some(Ok(_)) => continue, + Some(Err(e)) => return Err(format!("lecture du résultat : {e}")), + } + }; + let _ = ws.close(None).await; + #[derive(Deserialize)] + struct R { + #[serde(default)] + text: String, + } + let r: R = serde_json::from_slice(&brut) + .map_err(|e| format!("réponse illisible du serveur de dictée : {e}"))?; + if marqueur(&r.text) { + return Ok(String::new()); + } + Ok(r.text.trim().to_string()) +} + +// --------------------------------------------------------------------------- +// Archives +// --------------------------------------------------------------------------- + +/// `asrSafeRel` : retire le dossier de premier niveau, refuse ce qui sort +/// du dossier ; "" pour le dossier racine. +pub fn safe_rel(name: &str) -> Result { + let n = name.trim().replace('\\', "/"); + if n.is_empty() || n.starts_with('/') { + return Err(format!( + "entrée d'archive au chemin absolu : {}", + go_quote(name) + )); + } + let parts: Vec<&str> = n.trim_matches('/').split('/').collect(); + if parts.contains(&"..") { + return Err(format!( + "entrée d'archive qui remonte hors du dossier : {}", + go_quote(name) + )); + } + if parts.len() <= 1 { + return Ok(String::new()); + } + Ok(crate::paths::go_join(&parts[1..])) +} + +/// `%q` du Go pour une chaîne simple. +fn go_quote(s: &str) -> String { + serde_json::to_string(s).unwrap_or_default() +} + +/// `asrExtract` : `.tar.bz2` décompressé dans `dest.part`, renommé à la fin ; +/// seuls dossiers et fichiers ordinaires (pas de liens). +pub fn extract(r: R, dest: &Path) -> Result<(), String> { + let tmp = PathBuf::from(format!("{}.part", dest.display())); + remove_all(&tmp)?; + std::fs::create_dir_all(&tmp).map_err(|e| e.to_string())?; + let res = (|| -> Result<(), String> { + let mut ar = tar::Archive::new(bzip2::read::MultiBzDecoder::new(r)); + for entry in ar.entries().map_err(|e| e.to_string())? { + let mut e = entry.map_err(|e| e.to_string())?; + let name = String::from_utf8_lossy(&e.path_bytes()).into_owned(); + let rel = safe_rel(&name)?; + if rel.is_empty() { + continue; + } + let cible = tmp.join(&rel); + match e.header().entry_type() { + tar::EntryType::Directory => { + std::fs::create_dir_all(&cible).map_err(|e| e.to_string())?; + } + tar::EntryType::Regular => { + if let Some(p) = cible.parent() { + std::fs::create_dir_all(p).map_err(|e| e.to_string())?; + } + let mut f = std::fs::File::create(&cible).map_err(|e| e.to_string())?; + std::io::copy(&mut e, &mut f).map_err(|e| e.to_string())?; + } + _ => continue, + } + } + Ok(()) + })(); + if let Err(e) = res { + let _ = std::fs::remove_dir_all(&tmp); + return Err(e); + } + if let Err(e) = remove_all(dest) { + let _ = std::fs::remove_dir_all(&tmp); + return Err(e); + } + std::fs::rename(&tmp, dest).map_err(|e| e.to_string()) +} + +/// `os.RemoveAll` : absent n'est pas une erreur. +fn remove_all(p: &Path) -> Result<(), String> { + match std::fs::remove_dir_all(p) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => match std::fs::remove_file(p) { + Ok(()) => Ok(()), + Err(_) => Err(e.to_string()), + }, + } +} + +/// Lecteur bloquant alimenté par le téléchargement (canal tokio). +struct ChanReader { + rx: tokio::sync::mpsc::Receiver>, + cur: Bytes, +} + +impl Read for ChanReader { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + while self.cur.is_empty() { + match self.rx.blocking_recv() { + Some(Ok(b)) => self.cur = b, + Some(Err(e)) => return Err(std::io::Error::other(e)), + None => return Ok(0), + } + } + let n = buf.len().min(self.cur.len()); + buf[..n].copy_from_slice(&self.cur.split_to(n)); + Ok(n) + } +} + +// --------------------------------------------------------------------------- +// Supervision +// --------------------------------------------------------------------------- + +/// `ProcessState.String()` du Go, et l'explication d'une mort par signal. +pub fn exit_reason(st: ExitStatus) -> String { + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(sig) = st.signal() { + let nom = match sig { + 1 => "hangup".to_string(), + 2 => "interrupt".into(), + 3 => "quit".into(), + 4 => "illegal instruction".into(), + 6 => "aborted".into(), + 7 => "bus error".into(), + 8 => "floating point exception".into(), + 9 => "killed".into(), + 11 => "segmentation fault".into(), + 13 => "broken pipe".into(), + 15 => "terminated".into(), + n => format!("signal {n}"), + }; + return format!( + "signal: {nom} (binaire compilé pour un autre processeur : image à reconstruire)" + ); + } + } + match st.code() { + Some(c) => format!("exit status {c}"), + None => "arrêt sans erreur".into(), + } +} + +/// `time.Duration.String()` pour des durées entières en millisecondes. +fn go_duration(d: Duration) -> String { + let ms = d.as_millis(); + if ms < 1000 { + return format!("{ms}ms"); + } + let s = d.as_secs(); + let (h, m, s) = (s / 3600, (s / 60) % 60, s % 60); + if h > 0 { + format!("{h}h{m}m{s}s") + } else if m > 0 { + format!("{m}m{s}s") + } else { + format!("{s}s") + } +} + +type SharedChild = Arc>; + +#[derive(Default)] +struct Server { + child: Option, + port: u16, + cfg: DictateCfg, + since: Option, +} + +/// État du téléchargement d'un modèle. +#[derive(Debug, Clone, Default, Serialize)] +pub struct DlState { + pub id: String, + pub running: bool, + pub pct: i64, + #[serde(rename = "error")] + pub err: String, +} + +/// La dictée de ce lancement. +pub struct Dictation { + home: PathBuf, + model_base: String, + server_bin: Option, + idle: Duration, + ready: Duration, + srv: tokio::sync::Mutex, + /// Le processus en cours de démarrage, visible sans le verrou. + starting: Mutex>, + idle_gen: AtomicU64, + dl: Mutex, + client: reqwest::Client, +} + +fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + m.lock().unwrap_or_else(|e| e.into_inner()) +} + +impl Dictation { + pub fn new(home: &Path) -> Self { + Self::with(home, ASR_MODEL_BASE, None, ASR_IDLE, ASR_READY) + } + + /// Pour les tests : source des modèles, binaire du serveur, délais. + pub fn with( + home: &Path, + model_base: &str, + server_bin: Option, + idle: Duration, + ready: Duration, + ) -> Self { + crate::chat::engine::install_crypto(); + Dictation { + home: home.to_path_buf(), + model_base: model_base.into(), + server_bin, + idle, + ready, + srv: tokio::sync::Mutex::new(Server::default()), + starting: Mutex::new(None), + idle_gen: AtomicU64::new(0), + dl: Mutex::new(DlState::default()), + client: reqwest::Client::builder() + .connect_timeout(Duration::from_secs(20)) + .build() + .unwrap_or_default(), + } + } + + pub fn home(&self) -> &Path { + &self.home + } + + /// `asrServerBin`. + fn server_bin(&self) -> String { + if let Some(b) = &self.server_bin { + return b.clone(); + } + for p in [ + std::env::var("LOKI_ASR_SERVER").unwrap_or_default(), + "/usr/local/bin/sherpa-onnx-offline-websocket-server".into(), + ] { + if !p.is_empty() && std::fs::metadata(&p).is_ok() { + return p; + } + } + crate::paths::look_path("sherpa-onnx-offline-websocket-server").unwrap_or_default() + } + + /// `asrEnsure` : un serveur vivant pour ces réglages ; son port. + pub async fn ensure(self: &Arc, cfg: DictateCfg) -> Result { + let mut s = self.srv.lock().await; + if let Some(c) = &s.child + && s.cfg == cfg + && matches!(lock(c).try_wait(), Ok(None)) + { + self.touch(); + return Ok(s.port); + } + self.stop_locked(&mut s); + self.start_locked(&mut s, cfg).await + } + + async fn start_locked( + self: &Arc, + s: &mut Server, + cfg: DictateCfg, + ) -> Result { + if !model_present(&self.home, &cfg.model) { + return Err(format!( + "modèle de dictée absent : {} (à télécharger dans Paramètres → Dictée)", + cfg.model + )); + } + let bin = self.server_bin(); + if bin.is_empty() { + return Err("serveur de dictée introuvable (image à reconstruire)".into()); + } + let port = std::net::TcpListener::bind("127.0.0.1:0") + .and_then(|l| l.local_addr()) + .map(|a| a.port()) + .map_err(|e| format!("aucun port libre pour le serveur de dictée : {e}"))?; + let args = server_args(&self.home, &cfg, port)?; + let mut cmd = tokio::process::Command::new(&bin); + cmd.args(&args) + // Tous les GPU masqués (le binaire livré est CPU). + .env("CUDA_VISIBLE_DEVICES", "") + .env_remove(crate::auth::INTERNAL_ENV) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = cmd + .spawn() + .map_err(|e| format!("démarrage du serveur de dictée : {e}"))?; + let log = Arc::new(Mutex::new(Vec::::new())); + for out in [ + child + .stdout + .take() + .map(|o| Box::new(o) as Box), + child + .stderr + .take() + .map(|o| Box::new(o) as Box), + ] + .into_iter() + .flatten() + { + let log = log.clone(); + tokio::spawn(async move { + let mut out = out; + let mut buf = [0u8; 4096]; + while let Ok(n) = out.read(&mut buf).await { + if n == 0 { + break; + } + let mut l = lock(&log); + l.extend_from_slice(&buf[..n]); + // La fin seule renseigne. + let excess = l.len().saturating_sub(64 << 10); + l.drain(..excess); + } + }); + } + let child: SharedChild = Arc::new(Mutex::new(child)); + s.child = Some(child.clone()); + s.port = port; + s.cfg = cfg; + s.since = Some(Instant::now()); + *lock(&self.starting) = Some(child.clone()); + let res = self.wait_ready(&child, port, &log).await; + *lock(&self.starting) = None; + if let Err(e) = res { + self.stop_locked(s); + return Err(e); + } + self.touch(); + Ok(port) + } + + /// `asrAttendPret` : le serveur ouvre sa socket une fois le modèle chargé. + async fn wait_ready( + &self, + child: &SharedChild, + port: u16, + log: &Arc>>, + ) -> Result<(), String> { + let fin = Instant::now() + self.ready; + while Instant::now() < fin { + tokio::time::sleep(Duration::from_millis(200)).await; + let mort = lock(child).try_wait().ok().flatten(); + if let Some(st) = mort { + // Laisse les lecteurs du journal finir. + tokio::time::sleep(Duration::from_millis(50)).await; + let mut msg = exit_reason(st); + let l = last_line(&String::from_utf8_lossy(&lock(log))); + if !l.is_empty() { + msg += &format!(" — dernière sortie : {l}"); + } + return Err(format!( + "le serveur de dictée s'est arrêté au démarrage : {msg}" + )); + } + let conn = tokio::time::timeout( + Duration::from_secs(2), + tokio::net::TcpStream::connect(("127.0.0.1", port)), + ) + .await; + if matches!(conn, Ok(Ok(_))) { + return Ok(()); + } + } + Err(format!( + "le serveur de dictée n'a pas répondu en {}", + go_duration(self.ready) + )) + } + + /// `asrTouchLocked` : l'arrêt pour inactivité repart de zéro. + fn touch(self: &Arc) { + let tour = self.idle_gen.fetch_add(1, Ordering::SeqCst) + 1; + let me = Arc::downgrade(self); + let idle = self.idle; + tokio::spawn(async move { + tokio::time::sleep(idle).await; + if let Some(me) = me.upgrade() + && me.idle_gen.load(Ordering::SeqCst) == tour + { + me.shutdown().await; + } + }); + } + + fn stop_locked(&self, s: &mut Server) { + self.idle_gen.fetch_add(1, Ordering::SeqCst); + if let Some(c) = s.child.take() { + let _ = lock(&c).start_kill(); + } + *s = Server::default(); + } + + /// `asrShutdown` : inactivité, réglages changés. + pub async fn shutdown(&self) { + let mut s = self.srv.lock().await; + self.stop_locked(&mut s); + } + + /// `asrShutdownVite` : jamais derrière un démarrage (qui tient le verrou + /// jusqu'à deux minutes) — le processus qui démarre est tué. + pub async fn shutdown_fast(&self) { + for _ in 0..20 { + if let Ok(mut s) = self.srv.try_lock() { + self.stop_locked(&mut s); + return; + } + if let Some(c) = lock(&self.starting).as_ref() { + let _ = lock(c).start_kill(); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + /// `asrInfer`. + pub async fn infer(self: &Arc, cfg: DictateCfg, wav: &[u8]) -> Result { + let port = self.ensure(cfg).await?; + infer_at(&format!("ws://127.0.0.1:{port}"), wav).await + } + + /// `asrEtat`. + pub async fn etat(&self, cfg: &DictateCfg) -> Value { + let s = self.srv.lock().await; + let mut e = json!({ + "actif": s.child.is_some(), + "modele": cfg.model, + "present": model_present(&self.home, &cfg.model), + "device": "cpu", + }); + if s.child.is_some() + && let Some(t) = s.since + { + e["depuis_s"] = json!(t.elapsed().as_secs()); + } + e + } + + pub fn dl_state(&self) -> DlState { + lock(&self.dl).clone() + } + + fn clear_dl_err(&self) { + lock(&self.dl).err.clear(); + } + + /// `asrStartDownload` : un seul à la fois. + pub fn start_download(self: &Arc, id: &str) { + let mut d = lock(&self.dl); + if d.running { + return; + } + if model_url_for(&self.model_base, id).is_empty() { + d.err = format!("modèle de dictée inconnu : {id}"); + return; + } + *d = DlState { + id: id.into(), + running: true, + pct: 0, + err: String::new(), + }; + drop(d); + let me = self.clone(); + let id = id.to_string(); + tokio::spawn(async move { + let res = me.download(&id).await; + let mut d = lock(&me.dl); + d.running = false; + if let Err(e) = res { + d.err = e; + } + }); + } + + /// `asrDownload` : décompressé au fil du téléchargement ; progression + /// sur les octets reçus. + async fn download(self: &Arc, id: &str) -> Result<(), String> { + let dst = model_dir_for(&self.home, id) + .ok_or_else(|| format!("modèle de dictée inconnu : {id}"))?; + if let Some(p) = dst.parent() { + std::fs::create_dir_all(p).map_err(|e| e.to_string())?; + } + let mut resp = self + .client + .get(model_url_for(&self.model_base, id)) + .send() + .await + .map_err(|e| e.to_string())?; + if resp.status().as_u16() != 200 { + return Err(format!( + "téléchargement du modèle de dictée : HTTP {}", + resp.status().as_u16() + )); + } + let total = resp.content_length().unwrap_or(0) as i64; + let (tx, rx) = tokio::sync::mpsc::channel::>(32); + let extractor = tokio::task::spawn_blocking(move || { + extract( + ChanReader { + rx, + cur: Bytes::new(), + }, + &dst, + ) + }); + let mut lus = 0i64; + loop { + match resp.chunk().await { + Ok(Some(c)) => { + lus += c.len() as i64; + if total > 0 { + lock(&self.dl).pct = lus * 100 / total; + } + if tx.send(Ok(c)).await.is_err() { + break; // l'extraction a échoué : elle dira pourquoi + } + } + Ok(None) => break, + Err(e) => { + let _ = tx.send(Err(e.to_string())).await; + break; + } + } + } + drop(tx); + extractor.await.map_err(|e| e.to_string())? + } +} + +// --------------------------------------------------------------------------- +// Routes +// --------------------------------------------------------------------------- + +/// `dictateCfgLoad`. +pub async fn cfg_load(ctx: &Ctx) -> DictateCfg { + let raw = ctx + .store + .get(BK_STATE, DICTATE_CFG_KEY) + .await + .ok() + .flatten(); + raw.and_then(|b| serde_json::from_slice::(&b).ok()) + .unwrap_or_default() + .with_defauts() +} + +async fn read_body(body: Incoming, max: usize) -> Result { + Limited::new(body, max) + .collect() + .await + .map(|c| c.to_bytes()) + .map_err(|e| e.to_string()) +} + +fn err(code: StatusCode, msg: impl Into) -> Response { + send_json(code, &json!({ "error": msg.into() })) +} + +/// `from_go` : remise par le Go (tunnel), déjà authentifiée par lui. +pub async fn serve( + ctx: &Ctx, + d: &Arc, + req: Request, + from_go: bool, +) -> Response { + let (parts, body) = req.into_parts(); + if !from_go && let Err(refus) = ctx.guard(&parts.headers).await { + return *refus; + } + let post = parts.method == Method::POST; + match parts.uri.path() { + "/api/transcribe" => { + let audio = match read_body(body, 32 << 20).await { + Ok(a) if a.len() >= 1000 => a, + _ => return err(StatusCode::BAD_REQUEST, "audio vide ou trop court"), + }; + let cfg = cfg_load(ctx).await; + if !model_present(&d.home, &cfg.model) { + let st = d.dl_state(); + if !st.err.is_empty() { + // Échec précédent : on le dit ET on relance. + d.clear_dl_err(); + d.start_download(&cfg.model); + return send_json( + StatusCode::SERVICE_UNAVAILABLE, + &json!({ "downloading": true, "pct": 0, "error": st.err }), + ); + } + d.start_download(&cfg.model); + return send_json( + StatusCode::SERVICE_UNAVAILABLE, + &json!({ "downloading": true, "pct": st.pct }), + ); + } + match tokio::time::timeout(Duration::from_secs(300), d.infer(cfg, &audio)).await { + Ok(Ok(text)) => send_json(StatusCode::OK, &json!({ "text": text })), + Ok(Err(e)) => err(StatusCode::INTERNAL_SERVER_ERROR, e), + Err(_) => err( + StatusCode::INTERNAL_SERVER_ERROR, + "context deadline exceeded", + ), + } + } + "/api/dictate/config" if post => { + let c: DictateCfg = match read_body(body, 1 << 20) + .await + .ok() + .and_then(|b| serde_json::from_slice(&b).ok()) + { + Some(c) => c, + None => return err(StatusCode::BAD_REQUEST, "requête illisible"), + }; + if model(&c.model).is_none() { + return err( + StatusCode::BAD_REQUEST, + format!("modèle de dictée inconnu : {}", c.model), + ); + } + let raw = serde_json::to_vec(&c.with_defauts()).unwrap_or_default(); + if let Err(e) = ctx.store.put(BK_STATE, DICTATE_CFG_KEY, raw).await { + return err(StatusCode::BAD_REQUEST, e); + } + // Sans ça, l'ancien modèle continuerait de servir. + d.shutdown().await; + send_json(StatusCode::OK, &json!({ "ok": true })) + } + "/api/dictate/config" => send_json(StatusCode::OK, &cfg_load(ctx).await), + "/api/dictate/models" => { + let home = d.home.clone(); + let v = tokio::task::spawn_blocking(move || catalogue_trie(&home)) + .await + .unwrap_or(Value::Null); + send_json(StatusCode::OK, &v) + } + "/api/dictate/models/download" => { + if post { + #[derive(Deserialize)] + struct Req { + #[serde(default)] + id: String, + } + let r: Req = match read_body(body, 1 << 20) + .await + .ok() + .and_then(|b| serde_json::from_slice(&b).ok()) + { + Some(r) => r, + None => return err(StatusCode::BAD_REQUEST, "requête illisible"), + }; + let id = r.id.trim().to_string(); + if model_url_for(&d.model_base, &id).is_empty() { + return err( + StatusCode::BAD_REQUEST, + format!("modèle de dictée inconnu : {id}"), + ); + } + d.start_download(&id); + } + send_json(StatusCode::OK, &d.dl_state()) + } + "/api/dictate/state" => { + let cfg = cfg_load(ctx).await; + let mut e = d.etat(&cfg).await; + e["dl"] = json!(d.dl_state()); + send_json(StatusCode::OK, &e) + } + _ => send_json( + StatusCode::NOT_FOUND, + &json!({ "error": "404 page not found" }), + ), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn durees_comme_le_go() { + assert_eq!(go_duration(Duration::from_secs(120)), "2m0s"); + assert_eq!(go_duration(Duration::from_secs(1)), "1s"); + assert_eq!(go_duration(Duration::from_millis(500)), "500ms"); + assert_eq!(go_duration(Duration::from_secs(3725)), "1h2m5s"); + } + + #[test] + fn citations_comme_le_go() { + assert_eq!(go_quote("a\"b"), r#""a\"b""#); + } +} diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index c0f8b84..3da8474 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -15,6 +15,7 @@ pub mod auth; pub mod chat; pub mod chatapi; pub mod child; +pub mod dictate; pub mod downloads; pub mod engine; pub mod engstatus; diff --git a/rust/loki-rs/src/main.rs b/rust/loki-rs/src/main.rs index 7260207..fcd56b3 100644 --- a/rust/loki-rs/src/main.rs +++ b/rust/loki-rs/src/main.rs @@ -121,6 +121,9 @@ fn serve() -> ExitCode { // LOKI_MODELS=rust : les routes des modèles aussi (le Go remet alors // celles du tunnel et ne vide plus les .part). let models_rust = models_held_by_facade(); + // LOKI_DICTATE=rust : la dictée vocale aussi (le Go lui demande alors + // d'arrêter son serveur quand il libère la VRAM). + let dictate_rust = held_by_facade("LOKI_DICTATE"); let mut env = vec![(auth::INTERNAL_ENV.to_string(), token.clone())]; if chat_rust { env.push(("LOKI_CHAT".into(), "rust".into())); @@ -128,7 +131,10 @@ fn serve() -> ExitCode { if models_rust { env.push(("LOKI_MODELS".into(), "rust".into())); } - if chat_rust || models_rust { + if dictate_rust { + env.push(("LOKI_DICTATE".into(), "rust".into())); + } + if chat_rust || models_rust || dictate_rust { env.push(("LOKI_FACADE_ADDR".into(), format!("127.0.0.1:{web_port}"))); } let sup = child::Supervisor::spawn(go_bin, go_port, env); @@ -167,6 +173,11 @@ fn serve() -> ExitCode { None }; + let dictate = dictate_rust.then(|| { + eprintln!("[loki-rs] dictée tenue par la façade (LOKI_DICTATE=rust)"); + std::sync::Arc::new(loki_rs::dictate::Dictation::new(&engine_paths.home)) + }); + let addr = SocketAddr::from(([0, 0, 0, 0], web_port)); let listener = match tokio::net::TcpListener::bind(addr).await { Ok(l) => l, @@ -184,6 +195,7 @@ fn serve() -> ExitCode { engine: engine_paths, chat, models, + dictate, }; let res = proxy::serve(listener, cfg, shutdown_signal()).await; eprintln!("[loki-rs] arrêt demandé — arrêt du serveur Go"); @@ -205,7 +217,12 @@ fn chat_held_by_facade() -> bool { /// `LOKI_MODELS=rust` : routes des modèles tenues par la façade. fn models_held_by_facade() -> bool { - std::env::var("LOKI_MODELS").is_ok_and(|v| v.trim().eq_ignore_ascii_case("rust")) + held_by_facade("LOKI_MODELS") +} + +/// `=rust` : ce lot est tenu par la façade (le Go par défaut). +fn held_by_facade(var: &str) -> bool { + std::env::var(var).is_ok_and(|v| v.trim().eq_ignore_ascii_case("rust")) } /// Les `.part` d'un téléchargement tué ne reprendront jamais : vidés au diff --git a/rust/loki-rs/src/paths.rs b/rust/loki-rs/src/paths.rs index 6293938..b695929 100644 --- a/rust/loki-rs/src/paths.rs +++ b/rust/loki-rs/src/paths.rs @@ -396,6 +396,27 @@ pub fn base_name(p: &str) -> String { } } +/// `exec.LookPath` : le chemin du programme dans le PATH (exécutable). +pub fn look_path(name: &str) -> Option { + let path = std::env::var_os("PATH")?; + std::env::split_paths(&path) + .map(|dir| dir.join(name)) + .find(|p| { + std::fs::metadata(p).is_ok_and(|m| { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + m.is_file() && m.permissions().mode() & 0o111 != 0 + } + #[cfg(not(unix))] + { + m.is_file() + } + }) + }) + .map(|p| p.to_string_lossy().into_owned()) +} + /// Message d'une `*os.PathError` du Go : « op chemin: raison ». La raison /// suit la forme des `syscall.Errno` du Go sous Unix (minuscule en tête, sans /// le « (os error N) » de Rust). diff --git a/rust/loki-rs/src/proxy.rs b/rust/loki-rs/src/proxy.rs index 71c1076..ebc6839 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -66,6 +66,9 @@ pub struct Config { /// Routes des modèles tenues par la façade (`LOKI_MODELS=rust`) : servies /// ici ; `None` : relayées au Go. pub models: Option>, + /// Dictée vocale tenue par la façade (`LOKI_DICTATE=rust`) : servie ici ; + /// `None` : relayée au Go. + pub dictate: Option>, } struct Proxy { @@ -94,6 +97,7 @@ pub async fn serve( engine: cfg.engine, chat: cfg.chat, models: cfg.models, + dictate: cfg.dictate, internal_token: cfg.internal_token.clone(), }, }); @@ -133,10 +137,11 @@ pub async fn serve( impl Proxy { async fn handle(&self, mut req: Request) -> Response { // Le Go (et lui seul : jeton tiré à chaque démarrage) remet à la - // façade le trafic du chat et des modèles qui lui arrive sans passer - // par elle. - let from_go = (self.ctx.chat.is_some() || self.ctx.models.is_some()) - && self.ctx.from_go(req.headers()); + // façade le trafic du chat, des modèles et de la dictée qui lui arrive + // sans passer par elle. + let from_go = + (self.ctx.chat.is_some() || self.ctx.models.is_some() || self.ctx.dictate.is_some()) + && self.ctx.from_go(req.headers()); // Jamais de confiance accordée sur la foi d'un en-tête du client : // ni « je viens du tunnel », ni « je suis la façade ». req.headers_mut().remove(VIA_TUNNEL); @@ -150,6 +155,13 @@ impl Proxy { { return chatapi::serve_internal(chat, req).await; } + if from_go + && let Some(d) = &self.ctx.dictate + && path == crate::dictate::INTERNAL_STOP + { + d.shutdown_fast().await; + return plain(StatusCode::NO_CONTENT, ""); + } // L'API interne du Go n'est jamais joignable de l'extérieur. if path.starts_with("/internal/") { return plain(StatusCode::NOT_FOUND, "404 page not found"); @@ -169,6 +181,11 @@ impl Proxy { { return chatapi::serve(&self.ctx, chat, req, from_go).await; } + if let Some(d) = &self.ctx.dictate + && crate::dictate::handles(path) + { + return crate::dictate::serve(&self.ctx, d, req, from_go).await; + } if let Some(m) = &self.ctx.models && crate::models::handles(path) { diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 222bec9..bbddc8f 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -427,7 +427,7 @@ async fn handle(st: Arc, req: Request) -> Response /// Lance le faux Go puis la façade devant lui ; rend (adresse de la façade, /// état du faux Go, garde d'arrêt). pub async fn start(home: std::path::PathBuf) -> (SocketAddr, Arc, oneshot::Sender<()>) { - let (f, st, tx, _) = start_opts(home, false, None, None).await; + let (f, st, tx, _) = start_with(home, Opts::default()).await; (f, st, tx) } @@ -437,10 +437,36 @@ pub async fn start_models( home: std::path::PathBuf, hf: &str, ) -> (SocketAddr, Arc, oneshot::Sender<()>) { - let (f, st, tx, _) = start_opts(home, false, None, Some(hf.to_string())).await; + let opts = Opts { + models_hf: Some(hf.to_string()), + ..Opts::default() + }; + let (f, st, tx, _) = start_with(home, opts).await; (f, st, tx) } +/// Comme [`start`], avec la dictée tenue par la façade (`LOKI_DICTATE=rust`). +pub async fn start_dictate( + home: std::path::PathBuf, + d: Arc, +) -> (SocketAddr, Arc, oneshot::Sender<()>) { + let opts = Opts { + dictate: Some(d), + ..Opts::default() + }; + let (f, st, tx, _) = start_with(home, opts).await; + (f, st, tx) +} + +/// Ce que la façade tient en plus du proxy. +#[derive(Default)] +pub struct Opts { + pub chat: bool, + pub engine_port: Option, + pub models_hf: Option, + pub dictate: Option>, +} + /// Comme [`start`], avec la conversation tenue par la façade /// (`LOKI_CHAT=rust`) ; `engine_port` : port du faux moteur (config PORT). pub async fn start_chat( @@ -452,15 +478,21 @@ pub async fn start_chat( oneshot::Sender<()>, Arc, ) { - let (f, st, tx, chat) = start_opts(home, true, Some(engine_port), None).await; + let (f, st, tx, chat) = start_with( + home, + Opts { + chat: true, + engine_port: Some(engine_port), + ..Opts::default() + }, + ) + .await; (f, st, tx, chat.unwrap()) } -async fn start_opts( +pub async fn start_with( home: std::path::PathBuf, - with_chat: bool, - engine_port: Option, - models_hf: Option, + opts: Opts, ) -> ( SocketAddr, Arc, @@ -468,7 +500,7 @@ async fn start_opts( Option>, ) { let st = Arc::new(FakeGo::default()); - if let Some(p) = engine_port { + if let Some(p) = opts.engine_port { st.set("config", "PORT", &p.to_string()); } let l = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -489,7 +521,7 @@ async fn start_opts( }); } }); - let chat = if with_chat { + let chat = if opts.chat { use loki_rs::chat::{ conv::Conversation, generate::ChatService, host::ChatShared, sessions, }; @@ -517,12 +549,13 @@ async fn start_opts( service: "loki-engine".into(), }, chat: chat.clone(), - models: models_hf.map(|hf| { + models: opts.models_hf.map(|hf| { Arc::new(loki_rs::models::Models { downloads: Arc::new(loki_rs::downloads::Downloads::new()), hf: loki_rs::hf::HfClient::new(&hf), }) }), + dictate: opts.dictate, }; tokio::spawn(loki_rs::proxy::serve(fl, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/dictate.rs b/rust/loki-rs/tests/dictate.rs new file mode 100644 index 0000000..49bfe83 --- /dev/null +++ b/rust/loki-rs/tests/dictate.rs @@ -0,0 +1,705 @@ +//! Lot 9 : dictée vocale tenue par la façade (`LOKI_DICTATE=rust`). +//! +//! - le journal tiré du Go (`internal/loki/parity_dictate_test.go` → +//! `tests/parity_dictate/dictate.json`) : lecture des WAV, chemins +//! d'archive, réglages, catalogue, arguments du serveur, routes ; +//! - le protocole WebSocket de sherpa-onnx contre un faux serveur ; +//! - l'extraction d'une archive et le téléchargement d'un modèle ; +//! - la supervision d'un faux sherpa-onnx (script Python) par la façade. + +mod common; + +use std::convert::Infallible; +use std::io::Write; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bytes::Bytes; +use common::fakego; +use futures_util::{SinkExt, StreamExt}; +use http_body_util::Full; +use hyper::body::Incoming; +use hyper::{Request, Response, StatusCode, service::service_fn}; +use hyper_util::rt::TokioIo; +use loki_rs::dictate::{self, DictateCfg, Dictation}; +use serde_json::Value; +use tokio_tungstenite::tungstenite::Message; + +fn fixture() -> Value { + serde_json::from_str(include_str!("parity_dictate/dictate.json")).expect("journal") +} + +fn b64(s: &str) -> Vec { + const T: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = Vec::new(); + let mut acc = 0u32; + let mut n = 0; + for c in s.bytes().filter(|c| *c != b'=') { + acc = acc << 6 | T.iter().position(|t| *t == c).unwrap() as u32; + n += 6; + if n >= 8 { + n -= 8; + out.push((acc >> n) as u8); + acc &= (1 << n) - 1; + } + } + out +} + +/// v2 installé (quatre fichiers non vides), v3 à moitié — comme le journal. +fn install_models(home: &Path) { + for f in dictate::ASR_FICHIERS { + let p = dictate::model_dir_for(home, "parakeet-tdt-0.6b-v2") + .unwrap() + .join(f); + std::fs::create_dir_all(p.parent().unwrap()).unwrap(); + std::fs::write(p, b"x").unwrap(); + } + let v3 = dictate::model_dir_for(home, "parakeet-tdt-0.6b-v3").unwrap(); + std::fs::create_dir_all(&v3).unwrap(); + std::fs::write(v3.join("encoder.int8.onnx"), b"x").unwrap(); + std::fs::write(v3.join("tokens.txt"), b"").unwrap(); +} + +#[test] +fn parite_des_fonctions_pures() { + let fx = fixture(); + for w in fx["wav"].as_array().unwrap() { + let input = b64(w["in"].as_str().unwrap_or("")); + let name = w["name"].as_str().unwrap(); + match dictate::wav_to_f32(&input) { + Err(e) => assert_eq!(e, w["err"].as_str().unwrap_or(""), "{name}"), + Ok((rate, ech)) => { + assert!(w.get("err").is_none(), "{name} : erreur attendue"); + assert_eq!(rate as u64, w["rate"].as_u64().unwrap(), "{name}"); + let want: Vec = w + .get("ech") + .and_then(Value::as_array) + .map(|a| a.iter().map(|x| x.as_f64().unwrap() as f32).collect()) + .unwrap_or_default(); + assert_eq!(ech, want, "{name}"); + } + } + } + for c in fx["safe_rel"].as_array().unwrap() { + let input = c["in"].as_str().unwrap(); + match dictate::safe_rel(input) { + Ok(o) => assert_eq!(o, c["out"].as_str().unwrap_or(""), "{input:?}"), + Err(e) => assert_eq!(e, c["err"].as_str().unwrap_or(""), "{input:?}"), + } + } + for (s, want) in fx["marker"].as_object().unwrap() { + assert_eq!(dictate::marqueur(s), want.as_bool().unwrap(), "{s:?}"); + } + for c in fx["defaults"].as_array().unwrap() { + let d = DictateCfg { + model: c["model"].as_str().unwrap().into(), + reactivity: c["reactivity"].as_str().unwrap().into(), + } + .with_defauts(); + assert_eq!(d.model, c["out_model"].as_str().unwrap()); + assert_eq!(d.reactivity, c["out_reactivity"].as_str().unwrap()); + } + for (s, want) in fx["last_line"].as_object().unwrap() { + assert_eq!(dictate::last_line(s), want.as_str().unwrap(), "{s:?}"); + } + let home = PathBuf::from("/maison"); + let cfg = DictateCfg { + model: "parakeet-tdt-0.6b-v3".into(), + ..Default::default() + }; + let args: Vec = dictate::server_args(&home, &cfg, 4567) + .unwrap() + .iter() + .map(|a| a.replace("/maison", "{{HOME}}")) + .collect(); + let want: Vec = serde_json::from_value(fx["args"].clone()).unwrap(); + assert_eq!(args, want); + let bad = DictateCfg { + model: "x".into(), + ..Default::default() + }; + assert_eq!( + dictate::server_args(&home, &bad, 1).unwrap_err(), + fx["args_err"].as_str().unwrap() + ); + for (id, url) in fx["urls"].as_object().unwrap() { + assert_eq!( + dictate::model_url_for(dictate::ASR_MODEL_BASE, id), + url.as_str().unwrap(), + "{id}" + ); + } + let tmp = tempfile::tempdir().unwrap(); + install_models(tmp.path()); + for (id, want) in fx["present"].as_object().unwrap() { + assert_eq!( + dictate::model_present(tmp.path(), id), + want.as_bool().unwrap(), + "{id}" + ); + } +} + +#[tokio::test] +async fn parite_des_routes() { + let fx = fixture(); + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().canonicalize().unwrap(); + install_models(&home); + let d = Arc::new(Dictation::with( + &home, + "http://127.0.0.1:1/", + Some("/nexiste/pas".into()), + Duration::from_secs(60), + Duration::from_secs(5), + )); + let (facade, _go, _stop) = fakego::start_dictate(home.clone(), d).await; + let h = home.to_string_lossy().into_owned(); + for (i, c) in fx["calls"].as_array().unwrap().iter().enumerate() { + let (m, p) = (c["method"].as_str().unwrap(), c["path"].as_str().unwrap()); + let body = c.get("body").and_then(Value::as_str).unwrap_or(""); + let (code, b) = fakego::send(facade, m, p, &[], body).await; + let got: Value = + serde_json::from_str(&String::from_utf8_lossy(&b).replace(&h, "{{HOME}}")).unwrap(); + assert_eq!( + code.as_u16() as u64, + c["status"].as_u64().unwrap(), + "appel {i} : {m} {p} {body}" + ); + assert_eq!(got, c["out"], "appel {i} : {m} {p} {body}"); + } +} + +// --------------------------------------------------------------------------- +// Protocole +// --------------------------------------------------------------------------- + +fn wav16(n: usize) -> Vec { + let data: Vec = (0..n) + .flat_map(|i| ((i as i16).wrapping_mul(7)).to_le_bytes()) + .collect(); + let mut b = b"RIFF\0\0\0\0WAVEfmt ".to_vec(); + b.extend_from_slice(&16u32.to_le_bytes()); + b.extend_from_slice(&1u16.to_le_bytes()); + b.extend_from_slice(&1u16.to_le_bytes()); + b.extend_from_slice(&16000u32.to_le_bytes()); + b.extend_from_slice(&32000u32.to_le_bytes()); + b.extend_from_slice(&2u16.to_le_bytes()); + b.extend_from_slice(&16u16.to_le_bytes()); + b.extend_from_slice(b"data"); + b.extend_from_slice(&(data.len() as u32).to_le_bytes()); + b.extend_from_slice(&data); + let len = (b.len() - 8) as u32; + b[4..8].copy_from_slice(&len.to_le_bytes()); + b +} + +/// Faux sherpa-onnx : relit l'en-tête, attend le compte annoncé, répond. +async fn fake_ws(reply: &'static str) -> (String, Arc>>) { + let frames = Arc::new(Mutex::new(Vec::new())); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + let f = frames.clone(); + tokio::spawn(async move { + while let Ok((s, _)) = l.accept().await { + let f = f.clone(); + tokio::spawn(async move { + let Ok(mut ws) = tokio_tungstenite::accept_async(s).await else { + return; + }; + let mut buf = Vec::new(); + while let Some(Ok(m)) = ws.next().await { + if let Message::Binary(b) = m { + f.lock().unwrap().push(b.len()); + buf.extend_from_slice(&b); + if buf.len() >= 8 { + let n = u32::from_le_bytes(buf[4..8].try_into().unwrap()) as usize; + if buf.len() - 8 >= n { + let _ = ws.send(Message::Text(reply.into())).await; + return; + } + } + } + } + }); + } + }); + (format!("ws://{addr}"), frames) +} + +#[tokio::test] +async fn protocole_websocket() { + let (adr, frames) = fake_ws(r#"{"text":" bonjour ceci est un test "}"#).await; + // 6000 échantillons : 8 + 24000 octets → trois trames de 10 Kio au plus. + let txt = dictate::infer_at(&adr, &wav16(6000)).await.unwrap(); + assert_eq!(txt, "bonjour ceci est un test"); + assert_eq!(*frames.lock().unwrap(), [10240, 10240, 3528]); + + let (adr, _) = fake_ws(r#"{"text":" [musique] "}"#).await; + assert_eq!(dictate::infer_at(&adr, &wav16(10)).await.unwrap(), ""); + let (adr, _) = fake_ws("pas du json").await; + let e = dictate::infer_at(&adr, &wav16(10)).await.unwrap_err(); + assert!( + e.starts_with("réponse illisible du serveur de dictée : "), + "{e}" + ); + // Audio vide : rien à transcrire, aucune connexion. + assert_eq!( + dictate::infer_at("ws://127.0.0.1:1", &wav16(0)) + .await + .unwrap(), + "" + ); + let e = dictate::infer_at("ws://127.0.0.1:1", &wav16(10)) + .await + .unwrap_err(); + assert!(e.starts_with("connexion au serveur de dictée : "), "{e}"); + let e = dictate::infer_at("ws://127.0.0.1:1", b"pas un wav") + .await + .unwrap_err(); + assert_eq!(e, "audio illisible : ce n'est pas un WAV"); +} + +// --------------------------------------------------------------------------- +// Archives et téléchargement +// --------------------------------------------------------------------------- + +/// Une archive `.tar.bz2` de modèle : dossier racine, quatre fichiers, un +/// sous-dossier et un lien (ignoré). +fn archive(root: &str, evil: bool) -> Vec { + let enc = bzip2::write::BzEncoder::new(Vec::new(), bzip2::Compression::fast()); + let mut tb = tar::Builder::new(enc); + let mut dir = tar::Header::new_gnu(); + dir.set_entry_type(tar::EntryType::Directory); + dir.set_mode(0o755); + dir.set_size(0); + tb.append_data(&mut dir, format!("{root}/"), std::io::empty()) + .unwrap(); + for (i, f) in dictate::ASR_FICHIERS + .iter() + .chain(["sous/notes.txt"].iter()) + .enumerate() + { + let body = format!("contenu {i}"); + let mut h = tar::Header::new_gnu(); + h.set_mode(0o644); + h.set_size(body.len() as u64); + tb.append_data(&mut h, format!("{root}/{f}"), body.as_bytes()) + .unwrap(); + } + let mut l = tar::Header::new_gnu(); + l.set_entry_type(tar::EntryType::Symlink); + l.set_size(0); + tb.append_link(&mut l, format!("{root}/lien"), "/etc/passwd") + .unwrap(); + if evil { + let mut h = tar::Header::new_old(); + let name = format!("{root}/../evade.txt"); + h.as_old_mut().name[..name.len()].copy_from_slice(name.as_bytes()); + h.set_mode(0o644); + h.set_size(1); + h.set_cksum(); + tb.append(&h, &b"x"[..]).unwrap(); + } + let enc = tb.into_inner().unwrap(); + enc.finish().unwrap() +} + +#[test] +fn extraction_d_une_archive() { + let tmp = tempfile::tempdir().unwrap(); + let dest = tmp.path().join("asr/modele"); + std::fs::create_dir_all(dest.parent().unwrap()).unwrap(); + dictate::extract(&archive("racine", false)[..], &dest).unwrap(); + assert_eq!( + std::fs::read_to_string(dest.join("tokens.txt")).unwrap(), + "contenu 3" + ); + assert_eq!( + std::fs::read_to_string(dest.join("sous/notes.txt")).unwrap(), + "contenu 4" + ); + assert!(!dest.join("lien").exists(), "lien suivi"); + assert!(!tmp.path().join("asr/modele.part").exists()); + + // Entrée qui remonte : refus, rien de partiel ni d'évadé, l'ancien gardé. + let e = dictate::extract(&archive("racine", true)[..], &dest).unwrap_err(); + assert_eq!( + e, + r#"entrée d'archive qui remonte hors du dossier : "racine/../evade.txt""# + ); + assert!(!tmp.path().join("asr/modele.part").exists()); + assert!(!tmp.path().join("asr/evade.txt").exists()); + assert!(dest.join("tokens.txt").exists()); + // Archive corrompue. + assert!(dictate::extract(&b"pas du bzip2"[..], &dest).is_err()); + assert!(!tmp.path().join("asr/modele.part").exists()); +} + +async fn file_server(files: Vec<(String, Vec)>) -> SocketAddr { + let files = Arc::new(files); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((s, _)) = l.accept().await { + let files = files.clone(); + tokio::spawn(async move { + let svc = service_fn(move |req: Request| { + let files = files.clone(); + async move { + let p = req.uri().path().to_string(); + let r = match files.iter().find(|(n, _)| *n == p) { + Some((_, b)) => Response::new(Full::new(Bytes::from(b.clone()))), + None => { + let mut r = Response::new(Full::new(Bytes::from_static(b"absent"))); + *r.status_mut() = StatusCode::NOT_FOUND; + r + } + }; + Ok::<_, Infallible>(r) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(s), svc) + .await; + }); + } + }); + addr +} + +async fn wait_dl(d: &Dictation) -> dictate::DlState { + for _ in 0..500 { + let st = d.dl_state(); + if !st.running { + return st; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("téléchargement jamais terminé"); +} + +#[tokio::test] +async fn telechargement_d_un_modele() { + let dossier = "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8"; + let addr = file_server(vec![( + format!("/m/{dossier}.tar.bz2"), + archive(dossier, false), + )]) + .await; + let tmp = tempfile::tempdir().unwrap(); + let d = Arc::new(Dictation::with( + tmp.path(), + &format!("http://{addr}/m/"), + None, + Duration::from_secs(60), + Duration::from_secs(5), + )); + d.start_download("parakeet-tdt-0.6b-v2"); + // Un seul à la fois : le second est ignoré. + d.start_download("parakeet-tdt-0.6b-v3"); + let st = wait_dl(&d).await; + assert_eq!( + (st.id.as_str(), st.pct, st.err.as_str()), + ("parakeet-tdt-0.6b-v2", 100, "") + ); + assert!(dictate::model_present(tmp.path(), "parakeet-tdt-0.6b-v2")); + + d.start_download("parakeet-tdt-0.6b-v3"); + let st = wait_dl(&d).await; + assert_eq!(st.err, "téléchargement du modèle de dictée : HTTP 404"); + assert!(!dictate::model_present(tmp.path(), "parakeet-tdt-0.6b-v3")); + d.start_download("inconnu"); + assert_eq!(d.dl_state().err, "modèle de dictée inconnu : inconnu"); +} + +// --------------------------------------------------------------------------- +// Supervision d'un faux sherpa-onnx +// --------------------------------------------------------------------------- + +/// Faux `sherpa-onnx-offline-websocket-server` : poignée de main WebSocket, +/// lit l'utterance, répond ce qu'il a vu. +const FAKE_SHERPA: &str = r#"#!/usr/bin/env python3 +import base64, hashlib, json, os, socket, struct, sys +port = int([a for a in sys.argv if a.startswith('--port=')][0][7:]) +print("faux sherpa prêt", flush=True) +s = socket.socket(); s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +s.bind(('127.0.0.1', port)); s.listen(8) +def lire(c, n): + b = b'' + while len(b) < n: + x = c.recv(n - len(b)) + if not x: raise EOFError + b += x + return b +while True: + c, _ = s.accept() + try: + req = b'' + while b'\r\n\r\n' not in req: + x = c.recv(4096) + if not x: raise EOFError + req += x + key = [l.split(b':', 1)[1].strip() for l in req.split(b'\r\n') if l.lower().startswith(b'sec-websocket-key')][0] + acc = base64.b64encode(hashlib.sha1(key + b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11').digest()) + c.sendall(b'HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ' + acc + b'\r\n\r\n') + buf = b''; trames = 0 + while True: + h = lire(c, 2); op = h[0] & 15; n = h[1] & 127 + if n == 126: n = struct.unpack('>H', lire(c, 2))[0] + elif n == 127: n = struct.unpack('>Q', lire(c, 8))[0] + m = lire(c, 4); d = bytes(b ^ m[i % 4] for i, b in enumerate(lire(c, n))) + if op == 8: break + trames += 1; buf += d + if len(buf) >= 8 and len(buf) - 8 >= struct.unpack('H', len(t))) + t) + break + except Exception: + pass + finally: + c.close() +"#; + +fn script(dir: &Path, name: &str, body: &str) -> String { + let p = dir.join(name); + let mut f = std::fs::File::create(&p).unwrap(); + f.write_all(body.as_bytes()).unwrap(); + drop(f); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + p.to_string_lossy().into_owned() +} + +fn python3() -> bool { + std::process::Command::new("python3") + .arg("-c") + .arg("pass") + .status() + .is_ok_and(|s| s.success()) +} + +fn install_v3(home: &Path) { + for f in dictate::ASR_FICHIERS { + let p = dictate::model_dir_for(home, "parakeet-tdt-0.6b-v3") + .unwrap() + .join(f); + std::fs::create_dir_all(p.parent().unwrap()).unwrap(); + std::fs::write(p, b"x").unwrap(); + } +} + +async fn transcribe(facade: SocketAddr, wav: &[u8]) -> (StatusCode, Value) { + use http_body_util::BodyExt; + use hyper_util::client::legacy::Client; + use hyper_util::rt::TokioExecutor; + let client: Client<_, Full> = Client::builder(TokioExecutor::new()).build_http(); + let r = client + .request( + Request::post(format!("http://{facade}/api/transcribe")) + .body(Full::new(Bytes::copy_from_slice(wav))) + .unwrap(), + ) + .await + .unwrap(); + let code = r.status(); + let b = r.into_body().collect().await.unwrap().to_bytes(); + (code, serde_json::from_slice(&b).unwrap()) +} + +#[tokio::test] +async fn supervision_du_serveur_de_dictee() { + if !python3() { + eprintln!("python3 absent : test sauté"); + return; + } + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().canonicalize().unwrap(); + install_v3(&home); + let bin = script(&home, "faux-sherpa", FAKE_SHERPA); + let d = Arc::new(Dictation::with( + &home, + "http://127.0.0.1:1/", + Some(bin), + Duration::from_secs(60), + Duration::from_secs(20), + )); + let (facade, go, _stop) = fakego::start_dictate(home.clone(), d.clone()).await; + + // 3000 échantillons : 8 + 12000 octets → deux trames. GPU masqués, jeton + // interne absent de l'environnement du serveur. + let (code, v) = transcribe(facade, &wav16(3000)).await; + assert_eq!(code, StatusCode::OK, "{v}"); + assert_eq!( + v["text"], + "vu 3000 échantillons à 16000 Hz en 2 trames, cuda=[], jeton=False" + ); + let (_, _, b) = fakego::get(facade, "/api/dictate/state", &[]).await; + let st: Value = serde_json::from_slice(&b).unwrap(); + assert_eq!( + (st["actif"].as_bool(), st["present"].as_bool()), + (Some(true), Some(true)) + ); + assert!(st["depuis_s"].is_u64()); + // Même réglage : même serveur (pas de nouveau démarrage). + let port = d + .ensure(DictateCfg::default().with_defauts()) + .await + .unwrap(); + assert_eq!( + d.ensure(DictateCfg::default().with_defauts()) + .await + .unwrap(), + port + ); + + // Le Go demande l'arrêt (libération de la VRAM). + let (code, _) = fakego::send( + facade, + "POST", + dictate::INTERNAL_STOP, + &[("x-loki-internal", fakego::JETON)], + "{}", + ) + .await; + assert_eq!(code, StatusCode::NO_CONTENT); + let (_, _, b) = fakego::get(facade, "/api/dictate/state", &[]).await; + assert_eq!(serde_json::from_slice::(&b).unwrap()["actif"], false); + // Sans le jeton : la route interne n'est pas joignable. + let (code, _) = fakego::send(facade, "POST", dictate::INTERNAL_STOP, &[], "{}").await; + assert_eq!(code, StatusCode::NOT_FOUND); + + // Changer de réglage éteint le serveur. + let (code, _) = transcribe(facade, &wav16(1000)).await; + assert_eq!(code, StatusCode::OK); + let (code, _) = fakego::send( + facade, + "POST", + "/api/dictate/config", + &[], + r#"{"model":"parakeet-tdt-0.6b-v3","reactivity":"court"}"#, + ) + .await; + assert_eq!(code, StatusCode::OK); + assert_eq!( + go.get("state", "dictate_cfg").as_deref(), + Some(r#"{"model":"parakeet-tdt-0.6b-v3","reactivity":"court"}"#) + ); + let (_, _, b) = fakego::get(facade, "/api/dictate/state", &[]).await; + assert_eq!(serde_json::from_slice::(&b).unwrap()["actif"], false); +} + +#[tokio::test] +async fn demarrages_rates() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().canonicalize().unwrap(); + install_v3(&home); + let cfg = DictateCfg::default().with_defauts(); + // Mort au démarrage : la raison et la dernière ligne. + let mort = script( + &home, + "mort", + "#!/bin/sh\necho chargement\necho boom >&2\nexit 3\n", + ); + let d = Arc::new(Dictation::with( + &home, + "x", + Some(mort), + Duration::from_secs(60), + Duration::from_secs(10), + )); + let e = d.ensure(cfg.clone()).await.unwrap_err(); + assert_eq!( + e, + "le serveur de dictée s'est arrêté au démarrage : exit status 3 — dernière sortie : boom" + ); + // Muet : le délai, à la Go. + let muet = script(&home, "muet", "#!/bin/sh\nsleep 30\n"); + let d = Arc::new(Dictation::with( + &home, + "x", + Some(muet), + Duration::from_secs(60), + Duration::from_secs(1), + )); + let e = d.ensure(cfg.clone()).await.unwrap_err(); + assert_eq!(e, "le serveur de dictée n'a pas répondu en 1s"); + // Un arrêt « vite » pendant un démarrage tue le processus qui démarre. + let d2 = d.clone(); + let lent = Arc::new(Dictation::with( + &home, + "x", + Some(script(&home, "lent", "#!/bin/sh\nsleep 30\n")), + Duration::from_secs(60), + Duration::from_secs(60), + )); + let l2 = lent.clone(); + let c2 = cfg.clone(); + let h = tokio::spawn(async move { l2.ensure(c2).await }); + tokio::time::sleep(Duration::from_millis(300)).await; + let t0 = std::time::Instant::now(); + lent.shutdown_fast().await; + let e = tokio::time::timeout(Duration::from_secs(5), h) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!( + e.starts_with("le serveur de dictée s'est arrêté au démarrage : signal: killed"), + "{e}" + ); + assert!(t0.elapsed() < Duration::from_secs(4)); + drop(d2); + // Binaire introuvable, modèle absent. + let d = Arc::new(Dictation::with( + &home, + "x", + Some(String::new()), + Duration::from_secs(60), + Duration::from_secs(1), + )); + assert_eq!( + d.ensure(cfg).await.unwrap_err(), + "serveur de dictée introuvable (image à reconstruire)" + ); + let v2 = DictateCfg { + model: "parakeet-tdt-0.6b-v2".into(), + ..Default::default() + }; + assert_eq!( + d.ensure(v2).await.unwrap_err(), + "modèle de dictée absent : parakeet-tdt-0.6b-v2 (à télécharger dans Paramètres → Dictée)" + ); +} + +#[tokio::test] +async fn sans_la_variable_la_dictee_reste_au_go() { + let tmp = tempfile::tempdir().unwrap(); + let (facade, _go, _stop) = fakego::start(tmp.path().to_path_buf()).await; + for p in [ + "/api/transcribe", + "/api/dictate/config", + "/api/dictate/state", + ] { + let (_, _, b) = fakego::get(facade, p, &[]).await; + assert_eq!(&b[..], format!("go:{p}").as_bytes()); + } + let (code, _) = fakego::send( + facade, + "POST", + dictate::INTERNAL_STOP, + &[("x-loki-internal", fakego::JETON)], + "{}", + ) + .await; + assert_eq!(code, StatusCode::NOT_FOUND); +} diff --git a/rust/loki-rs/tests/lot2.rs b/rust/loki-rs/tests/lot2.rs index 349c12c..2c17ae7 100644 --- a/rust/loki-rs/tests/lot2.rs +++ b/rust/loki-rs/tests/lot2.rs @@ -95,6 +95,7 @@ async fn banc(cle: Option<&str>) -> Banc { }, chat: None, models: None, + dictate: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/parity_dictate/dictate.json b/rust/loki-rs/tests/parity_dictate/dictate.json new file mode 100644 index 0000000..93a5fcb --- /dev/null +++ b/rust/loki-rs/tests/parity_dictate/dictate.json @@ -0,0 +1,337 @@ +{ + "wav": [ + { + "name": "mono 16 bits", + "in": "UklGRi4AAABXQVZFZm10IBAAAAABAAEAgD4AAAB9AAACABAAZGF0YQoAAAAAAABAAMD/fwCA", + "rate": 16000, + "ech": [ + 0, + 0.5, + -0.5, + 0.9999695, + -1 + ] + }, + { + "name": "bloc LIST intercalé", + "in": "UklGRjgAAABXQVZFTElTVAcAAABJTkZPeHl6AGZtdCAQAAAAAQABAIA+AAAAfQAAAgAQAGRhdGEEAAAAZACc/w==", + "rate": 16000, + "ech": [ + 0.0030517578, + -0.0030517578 + ] + }, + { + "name": "stéréo moyenné", + "in": "UklGRiwAAABXQVZFZm10IBAAAAABAAIAgLsAAADuAgAEABAAZGF0YQgAAAAAAP9/AID/fw==", + "rate": 48000, + "ech": [ + 0.49998474, + -0.000015258789 + ] + }, + { + "name": "flottant 32 bits", + "in": "UklGRjAAAABXQVZFZm10IBAAAAADAAEAgD4AAAD6AAAEACAAZGF0YQwAAAAAAIA+AACAvwAAAD8=", + "rate": 16000, + "ech": [ + 0.25, + -1, + 0.5 + ] + }, + { + "name": "data tronqué", + "in": "UklGRioAAABXQVZFZm10IBAAAAABAAEAgD4AAAB9AAACABAAZGF0YQYAAAABAAIA", + "rate": 16000, + "ech": [ + 0.000030517578, + 0.000061035156 + ] + }, + { + "name": "data impair", + "in": "UklGRigAAABXQVZFZm10IBAAAAABAAEAQB8AAIA+AAACABAAZGF0YQMAAAABAgMA", + "rate": 8000, + "ech": [ + 0.015655518 + ] + }, + { + "name": "data vide", + "in": "UklGRiQAAABXQVZFZm10IBAAAAABAAEAgD4AAAB9AAACABAAZGF0YQAAAAA=", + "rate": 16000 + }, + { + "name": "24 bits", + "in": "UklGRioAAABXQVZFZm10IBAAAAABAAEAgD4AAIC7AAADABgAZGF0YQYAAAAAAAAAAAA=", + "err": "format audio non géré (format 1, 24 bits) — attendu PCM 16 bits" + }, + { + "name": "data avant fmt", + "in": "UklGRiYAAABXQVZFZGF0YQIAAAABAGZtdCAQAAAAAQABAIA+AAAAfQAAAgAQAA==", + "err": "audio illisible : bloc data avant fmt" + }, + { + "name": "fmt tronqué", + "in": "UklGRhQAAABXQVZFZm10IAgAAAAAAAAAAAAAAA==", + "err": "audio illisible : bloc fmt tronqué" + }, + { + "name": "sans data", + "in": "UklGRhwAAABXQVZFZm10IBAAAAABAAEAgD4AAAB9AAACABAA", + "err": "audio illisible : aucun bloc data" + }, + { + "name": "pas un WAV", + "in": "Y2VjaSBuJ2VzdCBwYXMgdW4gZmljaGllciBhdWRpbyBkdSB0b3V0", + "err": "audio illisible : ce n'est pas un WAV" + }, + { + "name": "vide", + "in": "", + "err": "audio illisible : ce n'est pas un WAV" + } + ], + "safe_rel": [ + { + "in": "", + "err": "entrée d'archive au chemin absolu : \"\"" + }, + { + "in": "racine" + }, + { + "in": "racine/" + }, + { + "in": "racine/encoder.onnx", + "out": "encoder.onnx" + }, + { + "in": "racine/sous/tokens.txt", + "out": "sous/tokens.txt" + }, + { + "in": "/abs/x", + "err": "entrée d'archive au chemin absolu : \"/abs/x\"" + }, + { + "in": "racine/../x", + "err": "entrée d'archive qui remonte hors du dossier : \"racine/../x\"" + }, + { + "in": "racine\\a\\b.txt", + "out": "a/b.txt" + }, + { + "in": " racine/a ", + "out": "a" + }, + { + "in": "..", + "err": "entrée d'archive qui remonte hors du dossier : \"..\"" + }, + { + "in": "a/b/../c", + "err": "entrée d'archive qui remonte hors du dossier : \"a/b/../c\"" + } + ], + "marker": { + "": true, + " ": true, + " (rires) ": true, + "(silence)": true, + "(x": false, + "[a] b": false, + "[musique]": true, + "bonjour": false + }, + "defaults": [ + { + "model": "", + "out_model": "parakeet-tdt-0.6b-v3", + "out_reactivity": "moyen", + "reactivity": "" + }, + { + "model": "ggml-base.bin", + "out_model": "parakeet-tdt-0.6b-v3", + "out_reactivity": "moyen", + "reactivity": "" + }, + { + "model": "parakeet-tdt-0.6b-v2", + "out_model": "parakeet-tdt-0.6b-v2", + "out_reactivity": "court", + "reactivity": "court" + } + ], + "last_line": { + "": "", + " \n x \n": "x", + "a\nb\n\n ": "b", + "un seul": "un seul" + }, + "args": [ + "--encoder={{HOME}}/asr/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8/encoder.int8.onnx", + "--decoder={{HOME}}/asr/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8/decoder.int8.onnx", + "--joiner={{HOME}}/asr/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8/joiner.int8.onnx", + "--tokens={{HOME}}/asr/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8/tokens.txt", + "--model-type=nemo_transducer", + "--port=4567", + "--host=127.0.0.1" + ], + "args_err": "modèle de dictée inconnu : x", + "urls": { + "inconnu": "", + "parakeet-tdt-0.6b-v2": "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8.tar.bz2", + "parakeet-tdt-0.6b-v3": "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8.tar.bz2" + }, + "calls": [ + { + "method": "GET", + "path": "/api/dictate/config", + "status": 200, + "out": { + "model": "parakeet-tdt-0.6b-v3", + "reactivity": "moyen" + } + }, + { + "method": "POST", + "path": "/api/dictate/config", + "body": "pas du json", + "status": 400, + "out": { + "error": "requête illisible" + } + }, + { + "method": "POST", + "path": "/api/dictate/config", + "body": "{\"model\":\"ggml-base.bin\"}", + "status": 400, + "out": { + "error": "modèle de dictée inconnu : ggml-base.bin" + } + }, + { + "method": "POST", + "path": "/api/dictate/config", + "body": "{\"model\":\"parakeet-tdt-0.6b-v2\"}", + "status": 200, + "out": { + "ok": true + } + }, + { + "method": "GET", + "path": "/api/dictate/config", + "status": 200, + "out": { + "model": "parakeet-tdt-0.6b-v2", + "reactivity": "moyen" + } + }, + { + "method": "POST", + "path": "/api/dictate/config", + "body": "{\"model\":\"parakeet-tdt-0.6b-v3\",\"reactivity\":\"long\",\"autre\":1}", + "status": 200, + "out": { + "ok": true + } + }, + { + "method": "GET", + "path": "/api/dictate/config", + "status": 200, + "out": { + "model": "parakeet-tdt-0.6b-v3", + "reactivity": "long" + } + }, + { + "method": "GET", + "path": "/api/dictate/models", + "status": 200, + "out": [ + { + "id": "parakeet-tdt-0.6b-v2", + "langues": "anglais uniquement", + "nom": "Parakeet v2 — anglais seul", + "octets": 482468385, + "present": true + }, + { + "id": "parakeet-tdt-0.6b-v3", + "langues": "25 langues européennes, français inclus — détection automatique", + "nom": "Parakeet v3 — recommandé", + "octets": 487170055, + "present": false + } + ] + }, + { + "method": "GET", + "path": "/api/dictate/models/download", + "status": 200, + "out": { + "error": "", + "id": "", + "pct": 0, + "running": false + } + }, + { + "method": "POST", + "path": "/api/dictate/models/download", + "body": "{\"id\":\" inconnu \"}", + "status": 400, + "out": { + "error": "modèle de dictée inconnu : inconnu" + } + }, + { + "method": "POST", + "path": "/api/dictate/models/download", + "body": "x", + "status": 400, + "out": { + "error": "requête illisible" + } + }, + { + "method": "GET", + "path": "/api/dictate/state", + "status": 200, + "out": { + "actif": false, + "device": "cpu", + "dl": { + "error": "", + "id": "", + "pct": 0, + "running": false + }, + "modele": "parakeet-tdt-0.6b-v3", + "present": false + } + }, + { + "method": "POST", + "path": "/api/transcribe", + "body": "trop court", + "status": 400, + "out": { + "error": "audio vide ou trop court" + } + } + ], + "present": { + "inconnu": false, + "parakeet-tdt-0.6b-v2": true, + "parakeet-tdt-0.6b-v3": false + } +} diff --git a/rust/loki-rs/tests/proxy.rs b/rust/loki-rs/tests/proxy.rs index bda1229..58b0b98 100644 --- a/rust/loki-rs/tests/proxy.rs +++ b/rust/loki-rs/tests/proxy.rs @@ -101,6 +101,7 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) { }, chat: None, models: None, + dictate: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await;