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;