diff --git a/README.md b/README.md index 4696595..d0e9501 100644 --- a/README.md +++ b/README.md @@ -55,12 +55,95 @@ Launch the web application: In debug mode: ``` FLASK_APP=shabda FLASK_DEBUG=1 flask run + +or + +FLASK_APP=shabda FLASK_DEBUG=1 poetry run flask run + ``` In production: ``` gunicorn --workers=4 "shabda:create_app()" -b localhost:8000 ``` +Then navigate to `http://localhost:5000/` (or your production URL). + +### Pack Tab + +Fetch random audio samples from Freesound based on words. + +- **Pack definition**: Enter one or more words separated by commas (e.g., `blue,red:3`). +- **Licenses**: Select which Creative Commons licenses to allow (cc0, by, by-nc). +- **Fetch pack**: Generates a reslist with sample URLs. + +Use in Estuary: +``` +!reslist "http://your-shabda-url/blue,red.json" +``` + +Use in Strudel: +``` +samples('shabda:blue:2,red:3') +``` + +### Speech Tab + +Generate Text-to-Speech samples from text input. + +- **Speech definition**: Enter words or sentences (spaces become underscores; `hello_world` produces a "hello world" sample). +- **Gender**: Choose Female or Male voice. +- **Language**: Select from 50+ languages (e.g., en-GB, en-US, fr-FR, ja-JP). +- **Fetch speech**: Generates audio samples using the selected voice. + +Use in Estuary: +``` +!reslist "http://your-shabda-url/speech/hello_world.json?language=en-GB&gender=f" +``` + +Use in Strudel: +``` +samples('shabda/speech/en-GB/f:hello_world') +``` + +### Phonemes Tab + +Generate phoneme-level samples with musical timing alignment. Each phoneme chunk becomes a sample, with stress-aware line wrapping for Strudel integration. + +**Inputs:** +- **Phoneme definition**: Enter lyrics (one or more sentences, any case). Commas and underscores separate phrases. Example: + ``` + Papa was a rolling stone + Wherever he laid his hat + ``` +- **ARPABET overrides**: Specify custom pronunciations for words in `word:PH1_PH2 format` (e.g., `papa:P_AA1_P_A;hat:HH_AE1_T`). Separate multiple overrides with semicolons. +- **Beats / bar**: Musical meter (default 4 for 4/4 time). +- **Bars / line**: Lines per phrase (default 2). +- **Target stress beat**: Which beat in the bar should primary stressed syllables land on (default 3 for a pickup feel; use 1 for downbeat emphasis). +- **Gender**: Choose Female or Male voice (default: Male). +- **Language**: Select pronunciation language (default: en-GB). + +**Actions:** +- **Preview phonemes**: Returns text analysis and Strudel timing without synthesizing audio (fast). +- **Fetch phonemes**: Synthesizes all phoneme samples as WAV files (slower first time, then cached). + +**Output includes:** +- IPA transcription of each word +- ARPABET phoneme breakdown +- Stress pattern analysis +- Timed Strudel phrases with beat alignment +- Direct audio sample URLs + +Use in Strudel: +``` +samples('shabda/phonemes/en-GB/m:papa_was_a_rolling_stone', { overrides: 'papa:P_AA1_P_A' }) +s("phc_P_AA1_P_A phc_W_AA1_Z phc_AH0 ~ phc_R_OW1_L_IH0_NG ...") +``` + +**How stress alignment works:** +- If a line's first phoneme chunk is stressed (marked with 1 or 2), it lands on beat 1 (no rest needed). +- If a line starts unstressed, rests are added so the first stressed chunk lands on the target beat (default beat 3). +- This ensures consistent musical phrasing across all lines. + Test ---- diff --git a/assets/shabda.css b/assets/shabda.css index 3446214..ff9a674 100644 --- a/assets/shabda.css +++ b/assets/shabda.css @@ -35,7 +35,8 @@ body { } #definition, -#speechdefinition { +#speechdefinition, +#phonemedefinition { border: 2px dotted #110072; border-radius: 5px; padding: 10px; @@ -44,8 +45,15 @@ body { color: #110072; } +#phonemedefinition { + min-height: 4.5em; + width: min(42rem, calc(100% - 20px)); + resize: vertical; +} + #definition:focus, -#speechdefinition:focus { +#speechdefinition:focus, +#phonemedefinition:focus { outline: none; border: 2px solid #110072; } @@ -69,6 +77,25 @@ body { margin-left: 10px; } +#timing { + display: flex; + flex-wrap: wrap; + gap: 12px; + align-items: center; + margin: 10px 0; +} + +#timing label { + display: inline-flex; + align-items: center; + gap: 6px; + font-size: 0.85em; +} + +#timing input[type=number] { + width: 5em; +} + #tabs span { display: inline-block; padding: 10px; @@ -98,7 +125,6 @@ button, margin: 10px; color: #110072; cursor: pointer; - background-color: white; text-decoration-line: none; font-size: 14px; } @@ -114,6 +140,31 @@ button:active, box-shadow: 1px 1px 2px #110072; } +#overrides { + margin: 8px 10px; +} + +#overrides label { + display: block; + font-size: 0.85em; + margin-bottom: 4px; +} + +#phoneme-overrides { + width: min(42rem, calc(100% - 20px)); + border: 2px dotted #110072; + border-radius: 5px; + padding: 10px; + margin: 10px; + background-color: transparent; + color: #110072; +} + +#phoneme-overrides:focus { + outline: none; + border: 2px solid #110072; +} + button img, .button img { vertical-align: top; @@ -135,6 +186,59 @@ button img, cursor: pointer; } +.phoneme-results { + display: block; +} + +.phoneme-sentences { + margin-top: 16px; +} + +.phoneme-sentence { + display: flex; + flex-wrap: wrap; + gap: 10px; + margin-bottom: 12px; +} + +.phoneme-word { + min-width: 10em; + padding: 10px 12px; + border: 1px solid #110072; + border-radius: 8px; + background-color: rgba(255, 255, 255, 0.5); +} + +.phoneme-word strong { + display: block; + margin-bottom: 4px; + font-size: 1rem; +} + +.phoneme-ipa, +.phoneme-arpabet, +.phoneme-stress { + display: block; + font-family: monospace; + font-size: 0.85em; + line-height: 1.3; +} + +.phoneme-strudel, +.phoneme-audio { + margin-top: 18px; +} + +.phoneme-bank { + margin-bottom: 12px; +} + +.phoneme-bank h4, +.phoneme-strudel h3, +.phoneme-audio h3 { + margin-bottom: 6px; +} + .copied { margin-left: 5px; opacity: 0; diff --git a/assets/shabda.js b/assets/shabda.js index f66bdb6..f5e361b 100644 --- a/assets/shabda.js +++ b/assets/shabda.js @@ -1,21 +1,171 @@ var root = document.body +var LANGUAGE_OPTIONS = [ + ["af-ZA", "Afrikaans"], + ["ar-XA", "Arabic"], + ["bn-IN", "Bengali"], + ["bg-BG", "Bulgarian"], + ["ca-ES", "Catalan"], + ["yue-HK", "Chinese (Hong Kong)"], + ["cs-CZ", "Czech"], + ["da-DK", "Danish"], + ["nl-BE", "Dutch (Belgium)"], + ["nl-NL", "Dutch (Netherlands)"], + ["en-AU", "English (Australia)"], + ["en-IN", "English (India)"], + ["en-GB", "English (UK)"], + ["en-US", "English (US)"], + ["fil-PH", "Filipino"], + ["fi-FI", "Finnish"], + ["fr-CA", "French (Canada)"], + ["fr-FR", "French (France)"], + ["de-DE", "German"], + ["el-GR", "Greek"], + ["gu-IN", "Gujarati"], + ["hi-IN", "Hindi"], + ["hu-HU", "Hungarian"], + ["is-IS", "Icelandic"], + ["id-ID", "Indonesian"], + ["it-IT", "Italian"], + ["ja-JP", "Japanese"], + ["kn-IN", "Kannada"], + ["ko-KR", "Korean"], + ["lv-LV", "Latvian"], + ["ms-MY", "Malay"], + ["ml-IN", "Malayalam"], + ["cmn-CN", "Mandarin Chinese"], + ["mr-IN", "Marathi"], + ["nb-NO", "Norwegian"], + ["pl-PL", "Polish"], + ["pt-BR", "Portuguese (Brazil)"], + ["pt-PT", "Portuguese (Portugal)"], + ["pa-IN", "Punjabi"], + ["ro-RO", "Romanian"], + ["ru-RU", "Russian"], + ["sr-RS", "Serbian"], + ["sk-SK", "Slovak"], + ["es-ES", "Spanish (Spain)"], + ["es-US", "Spanish (US)"], + ["sv-SE", "Swedish"], + ["ta-IN", "Tamil"], + ["te-IN", "Telugu"], + ["th-TH", "Thai"], + ["tr-TR", "Turkish"], + ["uk-UA", "Ukrainian"], + ["vi-VN", "Vietnamese"], +] + +function renderLanguageSelect(id, value, onChange) { + return m("select#" + id, { onchange: onChange, value: value }, + LANGUAGE_OPTIONS.map(function (option) { + return m("option[value=" + option[0] + "]", option[1]) + })) +} + +function renderCopyControls(type) { + return [ + m("small", "Insert in Estuary terminal: "), + m("pre", State.lastreslist(false, type)), + m("img", { onclick: function () { State.copyreslist(false, type) }, src: "assets/clipboard.png", height: "16" }), + State.copied ? m("span.copied", "copied") : null, + m("br"), + m("br"), + m("small", "Insert in Strudel: "), + m("pre", State.lastreslist(true, type)), + m("img", { onclick: function () { State.copyreslist(true, type) }, src: "assets/clipboard.png", height: "16" }), + State.strudelcopied ? m("span.copied", "copied") : null, + ] +} + +function renderPhonemeSentence(sentence) { + return m("div.phoneme-sentence", sentence.map(function (word) { + return m("div.phoneme-word", [ + m("strong", word.word), + word.ipa ? m("span.phoneme-ipa", word.ipa) : null, + word.arpabet ? m("span.phoneme-arpabet", word.arpabet) : null, + word.stress_pattern ? m("span.phoneme-stress", word.stress_pattern) : null, + ]) + })) +} + +function renderPhonemeBank(bank, urls) { + return m("div.phoneme-bank", [ + m("h4", bank), + m("ul.samples", urls.map(function (url) { + return m("li", m("audio", { controls: true, src: url })) + })) + ]) +} + +function flattenStringLists(nestedLists) { + var flat = [] + ;(nestedLists || []).forEach(function (list) { + ;(list || []).forEach(function (item) { + flat.push(String(item)) + }) + }) + return flat +} + +function toQuotedStringList(lines) { + if (!lines || lines.length === 0) { + return "[]" + } + return "[\n" + lines.map(function (line) { + return " \"" + String(line).replace(/\\/g, "\\\\").replace(/\"/g, "\\\"") + "\"," + }).join("\n") + "\n]" +} + +function normalizePhonemeDefinition(definition) { + return definition.trim() + .replace(/\r?\n+/g, ",") + .replace(/\s*,\s*/g, ",") + .replace(/[ \t]+/g, "_") + .toLowerCase() + .replace(/[^a-z0-9_,]+/g, "") + .replace(/_+/g, "_") + .replace(/,+/g, ",") + .replace(/^,|,$/g, "") +} + var State = { pack: "", - progress: false, - lastretrievedtype: "", - lastretrieved: "", - lastreslistcontents: [], - error: "", + results: { + pack: { progress: false, retrieved: "", contents: [], error: "" }, + speech: { progress: false, retrieved: "", contents: [], error: "" }, + phonemes: { progress: false, retrieved: "", contents: {}, error: "" }, + }, licenses: ["by", "cc0", "by-nc"], speech: "", - language: "uk-UA", - gender: "f", + speechLanguage: "uk-UA", + speechGender: "f", + phonemes: "",//Papa was a rolling stone\n(Wherever he laid his hat)\n(was his home)\n(And when he died)\n(All he left us was alone)\nSay Papa\nBuild Papa up\nScream Papa", + phonemeLanguage: "en-GB", + phonemeGender: "m", + phonemeOverrides: "", //"papa:P_AA1_P_A", + beatsPerBar: 4, + barsPerLine: 2, + targetStressBeat: 3, tab: "pack", + phonemePreview: true, freesoundAvailable: true, freesoundError: null, freesoundErrorReason: null, + getResult: function (type) { + return State.results[type] + }, + + currentType: function () { + if (State.tab == "speech") { + return "speech" + } + if (State.tab == "phonemes") { + return "phonemes" + } + return "pack" + }, + pollStatus: function () { m.request({ method: "GET", url: "/status" }) .then(function (result) { @@ -30,9 +180,9 @@ var State = { licensefullname: function (name) { switch (name) { - case 'cc0': return "Public domain" - case 'by': return "Attribution" - case 'by-nc': return "Attribution non-commercial" + case "cc0": return "Public domain" + case "by": return "Attribution" + case "by-nc": return "Attribution non-commercial" } }, @@ -41,90 +191,124 @@ var State = { }, setlicense: function (e) { - name = e.target.value + var name = e.target.value if (State.licenses.includes(name)) { - State.licenses = State.licenses.filter(function (value, index, arr) { - return value != name; - }); + State.licenses = State.licenses.filter(function (value) { + return value != name + }) } else { State.licenses.push(name) } }, - lastretrievedreslist: function () { - return this.lastretrieved + ".json" + lastretrievedreslist: function (type) { + var result = State.getResult(type) + return result.retrieved + ".json" }, - lastretrievedzip: function () { - return window.location + this.lastretrieved + ".zip" + lastretrievedzip: function (type) { + var result = State.getResult(type) + return window.location + result.retrieved + ".zip" + }, + + phonemezipurl: function () { + var result = State.getResult("phonemes") + if (!result.retrieved) { + return "" + } + var url = new URL(State.lastretrievedzip("phonemes")) + url.searchParams.append("gender", State.phonemeGender) + url.searchParams.append("language", State.phonemeLanguage) + if (State.phonemeOverrides.trim()) { + url.searchParams.append("overrides", State.phonemeOverrides.trim()) + } + return url.href }, - lastreslist: function (strudel = false) { - url = new URL(location.href + State.lastretrievedreslist()) - if (this.lastretrievedtype == "pack" && State.licenses.length < 3) { - url.searchParams.append("licenses", State.licenses.join()); + lastreslist: function (strudel = false, type = State.currentType()) { + var result = State.getResult(type) + if (!result.retrieved) { + return "" } - if (this.lastretrievedtype == "speech") { - url.searchParams.append("gender", State.gender) - url.searchParams.append("language", State.language) + var url = new URL(location.href + State.lastretrievedreslist(type)) + if (type == "pack" && State.licenses.length < 3) { + url.searchParams.append("licenses", State.licenses.join()) + } + if (type == "speech") { + url.searchParams.append("gender", State.speechGender) + url.searchParams.append("language", State.speechLanguage) + } + if (type == "phonemes") { + url.searchParams.append("gender", State.phonemeGender) + url.searchParams.append("language", State.phonemeLanguage) + if (State.phonemeOverrides.trim()) { + url.searchParams.append("overrides", State.phonemeOverrides.trim()) + } + url.searchParams.append("beats_per_bar", State.beatsPerBar) + url.searchParams.append("bars_per_line", State.barsPerLine) + url.searchParams.append("target_stress_beat", State.targetStressBeat) } if (strudel) { url.searchParams.append("strudel", 1) - reslist = "samples('" + url.href + "')" - } - else { - reslist = '!reslist "' + url.href + '"' + return "samples('" + url.href + "')" } - - return reslist + return '!reslist "' + url.href + '"' }, retrieve: function () { - if (State.pack) { - State.error = "" - State.progress = true - m.request({ - method: "GET", - url: "/pack/" + encodeURIComponent(State.pack) + '?licenses=' + State.licenses.join() - }) - .then(function (result) { - State.progress = false - if (result.status == "ok") { - State.lastretrieved = result.definition - State.lastretrievedtype = "pack" - m.request({ - method: "GET", - url: encodeURIComponent(State.lastretrievedreslist()) + '?complete=1&licenses=' + State.licenses.join() - }).then(function (result) { - State.lastreslistcontents = result - }) - } - else if (result.status == "empty") { - State.error = "Pack is empty" - } - else { - State.error = "An error occured" - } - }).catch(function (error) { - State.progress = false - if (error.code === 503 && error.response && error.response.status === "freesound_unavailable") { - State.error = "Freesound is currently unavailable. Try again later." - State.freesoundAvailable = false - State.freesoundError = error.response.error - } else { - State.error = "An error occured" - } - }) - } - else { - State.error = "Please enter a pack definition" + var result = State.getResult("pack") + if (!State.pack) { + result.error = "Please enter a pack definition" + return } + + result.error = "" + result.progress = true + m.request({ + method: "GET", + url: "/pack/" + encodeURIComponent(State.pack) + "?licenses=" + State.licenses.join(), + }) + .then(function (result) { + var packResult = State.getResult("pack") + packResult.progress = false + if (result.status == "ok") { + packResult.retrieved = result.definition + m.request({ + method: "GET", + url: encodeURIComponent(State.lastretrievedreslist("pack")) + "?complete=1&licenses=" + State.licenses.join(), + }).then(function (result) { + packResult.contents = result + }) + } + else if (result.status == "empty") { + packResult.error = "Pack is empty" + } + else { + packResult.error = "An error occured" + } + }) + .catch(function (error) { + var packResult = State.getResult("pack") + packResult.progress = false + if (error.code === 503 && error.response && error.response.status === "freesound_unavailable") { + packResult.error = "Freesound is currently unavailable. Try again later." + State.freesoundAvailable = false + State.freesoundError = error.response.error + } + else { + packResult.error = "An error occured" + } + }) }, - copyreslist: function (strudel = false) { + copyreslist: function (strudel = false, type = State.currentType()) { + var value = State.lastreslist(strudel, type) + if (!value) { + return + } if (navigator && navigator.clipboard && navigator.clipboard.writeText) { - if (navigator.clipboard.writeText(State.lastreslist(strudel))) { + if (navigator.clipboard.writeText(value)) { if (strudel) { State.strudelcopied = true } @@ -145,47 +329,113 @@ var State = { m.redraw() }, 2000) } - }, retrievespeech: function () { - if (State.speech) { - State.error = "" - State.progress = true - m.request({ - method: "GET", - url: "/speech/" + encodeURIComponent(State.speech) + '?language=' + State.language + '&gender=' + State.gender + var result = State.getResult("speech") + if (!State.speech) { + result.error = "Please enter a pack definition" + return + } + + result.error = "" + result.progress = true + m.request({ + method: "GET", + url: "/speech/" + encodeURIComponent(State.speech) + "?language=" + State.speechLanguage + "&gender=" + State.speechGender, + }) + .then(function (result) { + var speechResult = State.getResult("speech") + speechResult.progress = false + if (result.status == "ok") { + speechResult.retrieved = "speech/" + result.definition + m.request({ + method: "GET", + url: encodeURIComponent(State.lastretrievedreslist("speech")) + "?language=" + State.speechLanguage + "&gender=" + State.speechGender, + }).then(function (result) { + speechResult.contents = result + }) + } + else if (result.status == "empty") { + speechResult.error = "Pack is empty" + } + else { + speechResult.error = "An error occured" + } }) - .then(function (result) { - State.progress = false - if (result.status == "ok") { - State.lastretrieved = "speech/" + result.definition - State.lastretrievedtype = "speech" - m.request({ - method: "GET", - url: encodeURIComponent(State.lastretrievedreslist()) + '?language=' + State.language + '&gender=' + State.gender - }).then(function (result) { - State.lastreslistcontents = result - }) - } - else if (result.status == "empty") { - State.error = "Pack is empty" - } - else { - State.error = "An error occured" - } - }).catch(function (error) { - State.error = "An error occured" - }) + .catch(function () { + var speechResult = State.getResult("speech") + speechResult.progress = false + speechResult.error = "An error occured" + }) + }, + + retrievephonemes: function (preview = true) { + var result = State.getResult("phonemes") + if (!State.phonemes) { + result.error = "Please enter a phoneme definition" + return } - else { - State.error = "Please enter a pack definition" + + result.error = "" + result.progress = true + State.phonemePreview = preview + var phonemeDefinition = normalizePhonemeDefinition(State.phonemes) + var phonemeOverrides = State.phonemeOverrides.trim() + var query = + "?language=" + State.phonemeLanguage + + "&gender=" + State.phonemeGender + + "&beats_per_bar=" + State.beatsPerBar + + "&bars_per_line=" + State.barsPerLine + + "&target_stress_beat=" + State.targetStressBeat + + "&strudel=1&details=1&preview=" + (preview ? 1 : 0) + if (phonemeOverrides) { + query += "&overrides=" + encodeURIComponent(phonemeOverrides) } + m.request({ + method: "GET", + url: "/phonemes/" + encodeURIComponent(phonemeDefinition) + ".json" + query, + }) + .then(function (result) { + var phonemeResult = State.getResult("phonemes") + phonemeResult.progress = false + if (result._base) { + phonemeResult.retrieved = "phonemes/" + phonemeDefinition + phonemeResult.contents = result + } + else { + phonemeResult.error = "An error occured" + } + }) + .catch(function () { + var phonemeResult = State.getResult("phonemes") + phonemeResult.progress = false + phonemeResult.error = "An error occured" + }) }, } var Shabda = { view: function () { + var activeType = State.currentType() + var activeResult = State.getResult(activeType) + var activeContents = activeResult.contents || [] + var phonemeBanks = Object.keys(activeType == "phonemes" ? activeContents : {}).filter(function (key) { + if (key.charAt(0) == "_") { + return false + } + if ( + key == "sentences_strudel" || + key == "sentences_strudel_timed" || + key == "sentences_phonetic" || + key == "beats_per_bar" || + key == "bars_per_line" + ) { + return false + } + return Array.isArray(activeContents[key]) + }) + return m("main", [ !State.freesoundAvailable ? m("div#freesound-banner", State.freesoundErrorReason === "auth" @@ -201,7 +451,8 @@ var Shabda = { m("h1", [m("img", { src: "assets/logo.svg", height: "40", title: 'Shabda is the Sanskrit word for "speech sound"' }), "Shabda"]), - m("p.intro", ["Shabda is a tool for assembling and sharing packs of found audio samples.", + m("p.intro", [ + "Shabda is a tool for assembling and sharing packs of found audio samples.", m("br"), "It fetches samples from ", m("a", { href: "https://freesound.org/" }, "freesound.org"), @@ -211,25 +462,25 @@ var Shabda = { m("a", { href: "https://estuary.mcmaster.ca/" }, "Estuary"), " and ", m("a", { href: "https://strudel.tidalcycles.org/" }, "Strudel"), - "."]), + ".", + ]), - m("div", + m("div", [ m("div#tabs", [ - m("span#packs_tab", - { - class: State.tab == "pack" ? "selected" : "", - onclick: function () { State.tab = "pack" } - }, - "Pack" - ), - m("span#speech_tab", - { - class: State.tab == "speech" ? "selected" : "", - onclick: function () { State.tab = "speech" } - }, - "Speech" - ), + m("span#packs_tab", { + class: State.tab == "pack" ? "selected" : "", + onclick: function () { State.tab = "pack" }, + }, "Pack"), + m("span#speech_tab", { + class: State.tab == "speech" ? "selected" : "", + onclick: function () { State.tab = "speech" }, + }, "Speech"), + m("span#phonemes_tab", { + class: State.tab == "phonemes" ? "selected" : "", + onclick: function () { State.tab = "phonemes" }, + }, "Phonemes"), ]), + m("div.tabcontent", { style: State.tab == "pack" ? "display:block;" : "" }, [ m("input[placeholder=Pack definition]#definition", { value: State.pack, @@ -238,7 +489,7 @@ var Shabda = { if (e.keyCode == 13) { State.retrieve() } - } + }, }), m("div#licenses", [ m("input[type=checkbox]#license-cc0", { name: "licenses", value: "cc0", checked: State.haslicense("cc0"), oninput: State.setlicense }), @@ -250,16 +501,11 @@ var Shabda = { m("input[type=checkbox]#license-by-nc", { name: "licenses", value: "by-nc", checked: State.haslicense("by-nc"), oninput: State.setlicense }), m("label[for=license-by-nc]", State.licensefullname("by-nc")), ]), - m("button", { - onclick: State.retrieve - }, "Fetch pack"), - + m("button", { onclick: State.retrieve }, "Fetch pack"), m("br"), m("br"), - m("div.help", [ m("img", { src: "assets/help.png", height: "32", title: "Help" }), - m("p.explanation", [ "Any word can be a pack definition.", m("br"), @@ -274,9 +520,9 @@ var Shabda = { "In a hurry? You can directly include a pack in Estuary by executing in its terminal:", m("pre", '!reslist "' + location.href + 'blue,red.json"'), "Or in Strudel:", - m("pre", "samples('shabda:blue:2,red:3')") + m("pre", "samples('shabda:blue:2,red:3')"), ]), - ]) + ]), ]), m("div.clear"), @@ -289,80 +535,20 @@ var Shabda = { if (e.keyCode == 13) { State.retrievespeech() } - } + }, }), m("div#genders", [ - m("select#gender", { onchange: function (e) { State.gender = this.value } }, [ - m("option[value=f]", { selected: State.gender == "f" }, "Female"), - m("option[value=m]", { selected: State.gender == "m" }, "Male") + m("select#gender", { onchange: function () { State.speechGender = this.value } }, [ + m("option[value=f]", { selected: State.speechGender == "f" }, "Female"), + m("option[value=m]", { selected: State.speechGender == "m" }, "Male"), ]), ]), - m("div#languages", [ - m('select#language', { onchange: function (e) { State.language = this.value }, value: State.language }, [ - m("option[value=af-ZA]", "Afrikaans"), - m("option[value=ar-XA]", "Arabic"), - m("option[value=bn-IN]", "Bengali"), - m("option[value=bg-BG]", "Bulgarian"), - m("option[value=ca-ES]", "Catalan"), - m("option[value=yue-HK]", "Chinese (Hong Kong)"), - m("option[value=cs-CZ]", "Czech"), - m("option[value=da-DK]", "Danish"), - m("option[value=nl-BE]", "Dutch (Belgium)"), - m("option[value=nl-NL]", "Dutch (Netherlands)"), - m("option[value=en-AU]", "English (Australia)"), - m("option[value=en-IN]", "English (India)"), - m("option[value=en-GB]", "English (UK)"), - m("option[value=en-US]", "English (US)"), - m("option[value=fil-PH]", "Filipino"), - m("option[value=fi-FI]", "Finnish"), - m("option[value=fr-CA]", "French (Canada)"), - m("option[value=fr-FR]", "French (France)"), - m("option[value=de-DE]", "German"), - m("option[value=el-GR]", "Greek"), - m("option[value=gu-IN]", "Gujarati"), - m("option[value=hi-IN]", "Hindi"), - m("option[value=hu-HU]", "Hungarian"), - m("option[value=is-IS]", "Icelandic"), - m("option[value=id-ID]", "Indonesian"), - m("option[value=it-IT]", "Italian"), - m("option[value=ja-JP]", "Japanese"), - m("option[value=kn-IN]", "Kannada"), - m("option[value=ko-KR]", "Korean"), - m("option[value=lv-LV]", "Latvian"), - m("option[value=ms-MY]", "Malay"), - m("option[value=ml-IN]", "Malayalam"), - m("option[value=cmn-CN]", "Mandarin Chinese"), - m("option[value=mr-IN]", "Marathi"), - m("option[value=nb-NO]", "Norwegian"), - m("option[value=pl-PL]", "Polish"), - m("option[value=pt-BR]", "Portuguese (Brazil)"), - m("option[value=pt-PT]", "Portuguese (Portugal)"), - m("option[value=pa-IN]", "Punjabi"), - m("option[value=ro-RO]", "Romanian"), - m("option[value=ru-RU]", "Russian"), - m("option[value=sr-RS]", "Serbian"), - m("option[value=sk-SK]", "Slovak"), - m("option[value=es-ES]", "Spanish (Spain)"), - m("option[value=es-US]", "Spanish (US)"), - m("option[value=sv-SE]", "Swedish"), - m("option[value=ta-IN]", "Tamil"), - m("option[value=te-IN]", "Telugu"), - m("option[value=th-TH]", "Thai"), - m("option[value=tr-TR]", "Turkish"), - m("option[value=uk-UA]", "Ukrainian"), - m("option[value=vi-VN]", "Vietnamese") - ]) - ]), - m("button", { - onclick: State.retrievespeech - }, "Fetch speech"), - + m("div#languages", [renderLanguageSelect("language", State.speechLanguage, function () { State.speechLanguage = this.value })]), + m("button", { onclick: State.retrievespeech }, "Fetch speech"), m("br"), m("br"), - m("div.help", [ m("img", { src: "assets/help.png", height: "32", title: "Help" }), - m("p.explanation", [ "Any word can be a speech definition.", m("br"), @@ -377,63 +563,135 @@ var Shabda = { "In a hurry? You can directly include speech in Estuary by executing in its terminal:", m("pre", '!reslist "' + location.href + 'speech/blue,red.json?gender=f&language=en-GB"'), "Or in Strudel:", - m("pre", "samples('shabda/speech/en-GB/f:blue,red')") + m("pre", "samples('shabda/speech/en-GB/f:blue,red')"), ]), - ]) + ]), + ]), + + m("div.tabcontent", { style: State.tab == "phonemes" ? "display:block;" : "" }, [ + m("textarea[placeholder=Phoneme definition]#phonemedefinition", { + value: State.phonemes, + oninput: function (e) { State.phonemes = e.target.value }, + onkeyup: function (e) { + if (e.keyCode == 13) { + State.retrievephonemes() + } + }, + }), + m("div#timing", [ + m("label", ["Beats / bar ", m("input[type=number]#beatsperbar", { + min: 1, + step: 1, + value: State.beatsPerBar, + oninput: function (e) { State.beatsPerBar = Number(e.target.value || 4) }, + })]), + m("label", ["Bars / line ", m("input[type=number]#barsperline", { + min: 1, + step: 1, + value: State.barsPerLine, + oninput: function (e) { State.barsPerLine = Number(e.target.value || 2) }, + })]), + m("label", ["Target stress beat ", m("input[type=number]#targetstressbeat", { + min: 1, + step: 1, + value: State.targetStressBeat, + oninput: function (e) { State.targetStressBeat = Number(e.target.value || 3) }, + })]), + ]), + m("div#genders", [ + m("select#gender-phonemes", { onchange: function () { State.phonemeGender = this.value } }, [ + m("option[value=f]", { selected: State.phonemeGender == "f" }, "Female"), + m("option[value=m]", { selected: State.phonemeGender == "m" }, "Male"), + ]), + ]), + m("div#languages", [renderLanguageSelect("language-phonemes", State.phonemeLanguage, function () { State.phonemeLanguage = this.value })]), + m("div#overrides", [ + m("label[for=phoneme-overrides]", "ARPABET overrides"), + m("input#phoneme-overrides[placeholder=word:PH1_PH2;other:PH1_PH2]", { + value: State.phonemeOverrides, + oninput: function (e) { State.phonemeOverrides = e.target.value }, + }), + ]), + m("button", { onclick: function () { State.retrievephonemes(true) } }, "Preview phonemes"), + m("button", { onclick: function () { State.retrievephonemes(false) } }, "Fetch phonemes"), + m("br"), + m("br"), + m("div.help", [ + m("img", { src: "assets/help.png", height: "32", title: "Help" }), + m("p.explanation", [ + "Phoneme samples stay separated by word, while stress still shapes the sample boundaries.", + m("br"), + "Beat settings control how the Strudel phrase is padded with rests so each sample starts on a beat.", + m("br"), + "Preview first to get the text output immediately; generate audio when you want the WAV banks.", + m("br"), + "The result shows IPA, ARPABET, Strudel chunks, and the generated audio banks.", + m("br"), + m("br"), + "In a hurry? You can directly include phonemes in Strudel by executing in its terminal:", + m("pre", "samples('shabda/phonemes/en-GB/f:hello_world')"), + ]), + ]), ]), m("div.pack-container", [ - m("h2", "Samples"), + m("h2", activeType == "phonemes" ? "Phonemes" : "Samples"), m("div.pack", - State.error ? m("span.error", State.error) : - State.progress ? + activeResult.error ? m("span.error", activeResult.error) : + activeResult.progress ? m("img", { src: "assets/infinity.svg" }) : - State.lastretrieved ? - m("div.latest", [ - m("small", "Insert in Estuary terminal: "), - m("pre", State.lastreslist()), - m("img", { onclick: function () { State.copyreslist() }, src: "assets/clipboard.png", height: "16" }), - State.copied ? m("span.copied", "copied") : null, - m("br"), - m("br"), - m("small", "Insert in Strudel: "), - m("pre", State.lastreslist(true)), - m("img", { onclick: function () { State.copyreslist(true) }, src: "assets/clipboard.png", height: "16" }), - State.strudelcopied ? m("span.copied", "copied") : null, - m('ul.samples', function () { - soundlist = [] - for (index in State.lastreslistcontents) { - sound = State.lastreslistcontents[index] - soundlist.push( - m('li', - m('audio', { controls: true, src: sound.url }), - " ", - m('span.attribution', - sound.licensename ? State.licensefullname(sound.licensename) + ' by ' + sound.author + ' ' : '', - sound.original_url ? m('a', { 'href': sound.original_url }, m('img', { src: 'assets/external-link.png' })) : '' - ) + activeResult.retrieved ? + activeType == "phonemes" ? + m("div.latest.phoneme-results", [ + renderCopyControls("phonemes"), + m("div.phoneme-sentences", (activeContents.sentences_phonetic || []).map(renderPhonemeSentence)), + m("div.phoneme-strudel", [ + m("h3", "Timed Strudel lines"), + m("pre", m("code", toQuotedStringList(flattenStringLists(activeContents.sentences_strudel_timed)))), + m("h3", "Strudel chunks"), + m("pre", m("code", toQuotedStringList(activeContents.sentences_strudel || []))), + ]), + m("div.phoneme-audio", [ + m("h3", "Generated audio"), + phonemeBanks.length ? phonemeBanks.map(function (bank) { + return renderPhonemeBank(bank, activeContents[bank] || []) + }) : [ + m("p", "Preview mode does not synthesize audio banks yet."), + m("button", { onclick: function () { State.retrievephonemes(false) } }, "Fetch phonemes"), + ], + ]), + !State.phonemePreview && phonemeBanks.length + ? m("a.button", { href: State.phonemezipurl() }, "Download all") + : null, + ]) : + m("div.latest", [ + renderCopyControls(activeType), + m("ul.samples", activeContents.map(function (sound) { + return m("li", [ + m("audio", { controls: true, src: sound.url }), + " ", + m("span.attribution", + sound.licensename ? State.licensefullname(sound.licensename) + " by " + sound.author + " " : "", + sound.original_url ? m("a", { href: sound.original_url }, m("img", { src: "assets/external-link.png" })) : "" ) - ) - } - return soundlist - }()), - m("a.button", { href: State.lastretrievedzip() }, "Download all"), - ]) : - m("i", "Fetch a definition to hear its rendition") + ]) + })), + m("a.button", { href: State.lastretrievedzip(activeType) }, "Download all"), + ]) : + m("i", activeType == "phonemes" ? "Fetch a definition to inspect its phonemes" : "Fetch a definition to hear its rendition") ) ]), - ), - - m('p.footer', - [ - m('a', { href: "https://github.com/ilesinge/shabda", title: "Source code" }, m('img', { src: "assets/github.png", height: "16px" })), - ' by ', - m('a', { href: "https://mastodon.sdf.org/@detour" }, 'ilesinge') - ] - ) + ]), + + m("p.footer", [ + m("a", { href: "https://github.com/ilesinge/shabda", title: "Source code" }, m("img", { src: "assets/github.png", height: "16px" })), + " by ", + m("a", { href: "https://mastodon.sdf.org/@detour" }, "ilesinge"), + ]), ]) - } + }, } + m.mount(root, Shabda) State.pollStatus() setInterval(State.pollStatus, 30000) diff --git a/pyproject.toml b/pyproject.toml index 685b6d4..9e85704 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ flask = {extras = ["async"], version = "*"} gunicorn = "*" typer = {extras = ["all"], version = "*"} google-cloud-texttospeech = "*" +pronouncing = "*" [tool.poetry.group.dev.dependencies] pylint = "*" diff --git a/shabda/dj.py b/shabda/dj.py index ec2c52e..3068f06 100755 --- a/shabda/dj.py +++ b/shabda/dj.py @@ -9,6 +9,8 @@ import re import asyncio import urllib +import logging +import html import freesound import pydub @@ -22,6 +24,9 @@ import shabda.chatter as chatter +logger = logging.getLogger(__name__) + + class Dj: """Shabda engine""" @@ -106,13 +111,13 @@ def list( max_number, licenses=licenses, gender=gender, language=language ) - async def speak(self, word, language, gender): + async def speak(self, word, language, gender, force=False): """Speak a word""" # Only allow safe characters: letters, digits, underscore, dash language_sanitized = re.sub(r'[^\w\-]', '', language) sampleset = SampleSet(word, TTS, self.speech_samples_path) existing_samples = sampleset.list(language=language_sanitized, gender=gender) - if len(existing_samples) > 0: + if len(existing_samples) > 0 and not force: return True word_dir = sampleset.dir() client = texttospeech.TextToSpeechClient() @@ -158,6 +163,143 @@ async def speak(self, word, language, gender): sampleset.saveconfig() return True + async def speak_phoneme(self, arpabet_phone, ipa, language, gender, force=False): + """Synthesise a single phoneme via SSML and store as a TTS sample""" + key = "ph_" + arpabet_phone + sampleset = SampleSet(key, TTS, self.speech_samples_path) + existing_samples = sampleset.list(language=language, gender=gender) + if not force and any( + os.path.exists(sample.file) and os.path.getsize(sample.file) > 128 + for sample in existing_samples + ): + return True + word_dir = sampleset.dir() + client = texttospeech.TextToSpeechClient() + ssml = f'x' + logger.warning( + "[tts:ssml:phoneme] key=%s language=%s gender=%s ssml=%s", + key, + language, + gender, + ssml, + ) + synthesis_input = texttospeech.SynthesisInput(ssml=ssml) + + if gender == "f": + ssml_gender = texttospeech.SsmlVoiceGender.FEMALE + else: + ssml_gender = texttospeech.SsmlVoiceGender.MALE + + voice_name = chatter.pick_voice(language, ssml_gender, client) + + if language == "en-GB" and gender == "f": + voice = texttospeech.VoiceSelectionParams( + name="en-GB-Neural2-A", + language_code="en-GB", + ssml_gender=texttospeech.SsmlVoiceGender.FEMALE, + ) + else: + voice = texttospeech.VoiceSelectionParams( + name=voice_name, language_code=language + ) + audio_config = texttospeech.AudioConfig( + audio_encoding=texttospeech.AudioEncoding.LINEAR16, + ) + response = client.synthesize_speech( + input=synthesis_input, voice=voice, audio_config=audio_config + ) + filepath = word_dir + "/" + key + "_" + language + "_" + gender + ".wav" + with open(filepath, "wb") as out: + out.write(response.audio_content) + if len(existing_samples) > 0: + return True + sound = Sound( + speechsound={ + "gender": gender, + "language": language, + "file": filepath, + } + ) + sampleset.add(sound) + sampleset.saveconfig() + return True + + async def speak_phoneme_chunk( + self, + chunk_key, + ipa_phones, + language, + gender, + text_hint="", + force=False, + ): + """Synthesise a sequence of phonemes via SSML and store as a TTS sample""" + sampleset = SampleSet(chunk_key, TTS, self.speech_samples_path) + existing_samples = sampleset.list(language=language, gender=gender) + if not force and any( + os.path.exists(sample.file) and os.path.getsize(sample.file) > 128 + for sample in existing_samples + ): + return True + + word_dir = sampleset.dir() + client = texttospeech.TextToSpeechClient() + ipa_string = "".join(ipa_phones) + phoneme_body = html.escape(text_hint.strip() or "x") + ssml = ( + f'{phoneme_body}' + ) + logger.warning( + "[tts:ssml:chunk] key=%s language=%s gender=%s ssml=%s", + chunk_key, + language, + gender, + ssml, + ) + synthesis_input = texttospeech.SynthesisInput(ssml=ssml) + + if gender == "f": + ssml_gender = texttospeech.SsmlVoiceGender.FEMALE + else: + ssml_gender = texttospeech.SsmlVoiceGender.MALE + + voice_name = chatter.pick_voice(language, ssml_gender, client) + + if language == "en-GB" and gender == "f": + voice = texttospeech.VoiceSelectionParams( + name="en-GB-Neural2-A", + language_code="en-GB", + ssml_gender=texttospeech.SsmlVoiceGender.FEMALE, + ) + else: + voice = texttospeech.VoiceSelectionParams( + name=voice_name, language_code=language + ) + + audio_config = texttospeech.AudioConfig( + audio_encoding=texttospeech.AudioEncoding.LINEAR16, + ) + response = client.synthesize_speech( + input=synthesis_input, voice=voice, audio_config=audio_config + ) + + filepath = word_dir + "/" + chunk_key + "_" + language + "_" + gender + ".wav" + with open(filepath, "wb") as out: + out.write(response.audio_content) + if len(existing_samples) > 0: + return True + + sound = Sound( + speechsound={ + "gender": gender, + "language": language, + "file": filepath, + } + ) + sampleset.add(sound) + sampleset.saveconfig() + return True + async def fetch(self, word, num, licenses): """Fetch a collection of samples""" mastersound = None diff --git a/shabda/phonemize.py b/shabda/phonemize.py new file mode 100644 index 0000000..f4d288e --- /dev/null +++ b/shabda/phonemize.py @@ -0,0 +1,204 @@ +"""ARPABET phoneme decomposition and IPA conversion""" + +from functools import lru_cache + +import pronouncing +import re + +# Maps CMU ARPABET base phone (no stress digit) → IPA symbol. +# Source: CMU Pronouncing Dictionary phone set documentation. +ARPABET_TO_IPA = { + # Vowels + "AA": "ɑ", + "AE": "æ", + "AH": "ə", # unstressed (0); stressed (1/2) → "ʌ" handled below + "AO": "ɔ", + "AW": "aʊ", + "AY": "aɪ", + "EH": "ɛ", + "ER": "ɝ", + "EY": "eɪ", + "IH": "ɪ", + "IY": "iː", + "OW": "oʊ", + "OY": "ɔɪ", + "UH": "ʊ", + "UW": "uː", + # Consonants + "B": "b", + "CH": "tʃ", + "D": "d", + "DH": "ð", + "F": "f", + "G": "ɡ", + "HH": "h", + "JH": "dʒ", + "K": "k", + "L": "l", + "M": "m", + "N": "n", + "NG": "ŋ", + "P": "p", + "R": "ɹ", + "S": "s", + "SH": "ʃ", + "T": "t", + "TH": "θ", + "V": "v", + "W": "w", + "Y": "j", + "Z": "z", + "ZH": "ʒ", +} + +# AH with primary/secondary stress is "ʌ" (STRUT), not schwa +_AH_STRESSED_IPA = "ʌ" + + +@lru_cache(maxsize=4096) +def _phones_for_word(word: str) -> tuple[str, ...]: + """Cache CMU pronunciation lookups for repeated words.""" + phones_list = pronouncing.phones_for_word(word) + if phones_list: + return tuple(phones_list[0].split()) + return () + + +def sentence_to_arpabet( + sentence: str, + overrides: dict[str, list[str]] | None = None, +) -> list[tuple[str, list[str] | None]]: + """ + Break a sentence (underscores as spaces) into per-word ARPABET phone lists. + + Returns a list of (word, phones) tuples. For words found in the CMU + Pronouncing Dictionary, `phones` is a list of ARPABET tokens including + stress digits (e.g. ["HH", "AH0", "L", "OW1"]). For out-of-vocabulary + words, `phones` is None — callers should fall back to whole-word synthesis. + """ + # Treat punctuation and commas as separators so phrase definitions like + # "foo,bar_baz" are tokenized into clean words. + words = re.findall(r"[A-Za-z0-9]+", sentence.replace("_", " ")) + result = [] + overrides = overrides or {} + for word in words: + override_phones = overrides.get(word.lower()) + if override_phones: + result.append((word, override_phones)) + continue + phones = list(_phones_for_word(word.lower())) + if phones: + result.append((word, phones)) + else: + result.append((word, None)) + return result + + +def arpabet_to_ipa(phone: str) -> str: + """ + Convert a single ARPABET token (with optional stress digit) to an IPA string. + + Stress digits: + 0 → no prefix (unstressed) + 1 → ˈ (primary stress prefix) + 2 → ˌ (secondary stress prefix) + + Examples: + "AH0" → "ə" + "AH1" → "ˈʌ" + "OW1" → "ˈoʊ" + "HH" → "h" + """ + stress = "" + base = phone + if phone and phone[-1].isdigit(): + digit = phone[-1] + base = phone[:-1] + if digit == "1": + stress = "ˈ" + elif digit == "2": + stress = "ˌ" + # digit == "0" → no prefix + + # Special case: stressed AH (1 or 2) is STRUT vowel "ʌ", not schwa + if base == "AH" and stress: + ipa = _AH_STRESSED_IPA + else: + ipa = ARPABET_TO_IPA.get(base, base.lower()) + + return stress + ipa + + +def is_stressed_phone(phone: str) -> bool: + """Return True for ARPABET tokens marked with primary/secondary stress.""" + return bool(phone and phone[-1] in ("1", "2")) + + +def is_primary_stressed_phone(phone: str) -> bool: + """Return True for ARPABET tokens marked with primary stress only.""" + return bool(phone and phone[-1] == "1") + + +def _chunk_phones_pre_stress( + phones: list[str], + is_chunk_boundary, +) -> list[list[str]]: + """ + Split a phone sequence so each chunk ends right before a stressed phone. + + Example: + ["P", "AH0", "P", "AA1"] -> [["P", "AH0", "P"], ["AA1"]] + + This creates larger, beat-alignable chunks where stressed material starts + the following chunk. + """ + if not phones: + return [] + + chunks = [] + start = 0 + + for i, phone in enumerate(phones): + if i > start and is_chunk_boundary(phone): + chunks.append(phones[start:i]) + start = i + + chunks.append(phones[start:]) + chunks = [chunk for chunk in chunks if chunk] + + # Keep syllable onsets together with stressed nuclei by moving trailing + # consonants from the previous chunk to the next chunk when the next chunk + # starts on a stressed vowel (e.g. ["P"], ["AA1", "P"] -> ["P", "AA1", "P"]). + for i in range(1, len(chunks)): + curr = chunks[i] + prev = chunks[i - 1] + if not curr or not prev: + continue + if not is_chunk_boundary(curr[0]): + continue + + move_start = len(prev) + while move_start > 0 and not prev[move_start - 1][-1:].isdigit(): + move_start -= 1 + + if move_start < len(prev): + onset = prev[move_start:] + chunks[i - 1] = prev[:move_start] + chunks[i] = onset + curr + + return [chunk for chunk in chunks if chunk] + + +def chunk_phones_pre_primary_stress(phones: list[str]) -> list[list[str]]: + """Split phones right before primary stress only (digit 1).""" + return _chunk_phones_pre_stress(phones, is_primary_stressed_phone) + + +def chunk_phones_pre_any_stress(phones: list[str]) -> list[list[str]]: + """Split phones right before primary or secondary stress (digits 1/2).""" + return _chunk_phones_pre_stress(phones, is_stressed_phone) + + +def chunk_phones_pre_stress(phones: list[str]) -> list[list[str]]: + """Default chunking: split phones right before primary stress only.""" + return chunk_phones_pre_primary_stress(phones) diff --git a/shabda/templates/home.html b/shabda/templates/home.html index b45c1f7..420d1c4 100644 --- a/shabda/templates/home.html +++ b/shabda/templates/home.html @@ -5,17 +5,17 @@ Shabda - + - + - - + + \ No newline at end of file diff --git a/shabda/web.py b/shabda/web.py index fd17171..17a2a2a 100644 --- a/shabda/web.py +++ b/shabda/web.py @@ -4,6 +4,8 @@ import os from urllib.parse import urlparse import json +import re +import pronouncing from zipfile import ZipFile import tempfile @@ -20,6 +22,7 @@ from werkzeug.utils import secure_filename from shabda.dj import Dj from shabda.client import FreesoundUnavailableError +import shabda.phonemize as phonemize SHABDA_PATH = os.path.expanduser("~/.shabda/") @@ -30,6 +33,219 @@ bp = Blueprint("web", __name__, url_prefix="/") dj = Dj(SHABDA_PATH, SAMPLES_PATH, SPEECH_SAMPLE_PATH) +_phoneme_response_cache = {} + + +def safe_bank_name(name: str, preserve_underscores: bool = False) -> str: + """Normalize a token into a bank-safe key. + + Words use hyphens, while phone sequence keys can preserve underscores. + """ + name = name.strip() + if preserve_underscores: + name = re.sub(r"\s+", "-", name) + name = re.sub(r"[^A-Za-z0-9_-]+", "-", name) + name = re.sub(r"_+", "_", name) + name = re.sub(r"-+", "-", name) + return name.strip("-_") + + name = re.sub(r"[\s_]+", "-", name) + name = re.sub(r"[^A-Za-z0-9-]+", "-", name) + name = re.sub(r"-+", "-", name) + return name.strip("-") + + +def parse_arpabet_overrides(raw: str | None) -> dict[str, list[str]]: + """Parse ARPABET overrides from query string. + + Format: + overrides=word:PH1_PH2_PH3;other:PH1 PH2 + + Notes: + - entries are separated by ';' + - phones may be separated by '_' or whitespace + - words are matched case-insensitively + """ + if not raw: + return {} + + overrides = {} + entries = [entry.strip() for entry in raw.split(";") if entry.strip()] + for entry in entries: + if ":" not in entry: + continue + word, phones_raw = entry.split(":", 1) + word = "".join(ch for ch in word.lower().strip() if ch.isalnum()) + if not word: + continue + + phones = [ + token.upper() + for token in re.split(r"[_\s]+", phones_raw.strip()) + if token.strip() + ] + if not phones: + continue + + overrides[word] = phones + + return overrides + + +def sentence_chunk_plan(sentence_word_phones): + """Build a chunk plan where word boundaries remain sample boundaries.""" + plan = [] + for word, phones in sentence_word_phones: + if not phones: + plan.append(("oov", word)) + continue + + # Keep each word isolated so short unstressed words like "a" remain + # their own sample, while stressed syllables still begin the chunk. + for chunk in phonemize.chunk_phones_pre_stress(phones): + chunk_key_raw = "phc_" + "_".join(chunk) + plan.append(("chunk", chunk, word, chunk_key_raw)) + + return plan + + +def sentence_tokens_to_timed_lines( + tokens: list[str], + beats_per_bar: int = 4, + bars_per_line: int = 2, + lead_in_beats: int | None = None, + target_stress_beat: int = 3, +) -> list[str]: + """Pad tokens into fixed-size lines and place stressed chunks on strong beats.""" + beats_per_bar = max(1, beats_per_bar) + bars_per_line = max(1, bars_per_line) + target_stress_beat = max(1, min(target_stress_beat, beats_per_bar)) + slots_per_line = beats_per_bar * bars_per_line + + if not tokens: + return [" ".join(["~"] * slots_per_line)] + + def is_stressed_token(token: str) -> bool: + return bool(re.search(r"[A-Z]+[12](?:_|$)", token)) + + # Strong beats per bar: downbeat and (for 4/4-like meters) mid-bar accent. + strong_offsets = [0] + if beats_per_bar >= 4 and beats_per_bar % 2 == 0: + strong_offsets.append(beats_per_bar // 2) + + strong_slots = [] + for bar in range(bars_per_line): + bar_start = bar * beats_per_bar + for offset in strong_offsets: + slot = bar_start + offset + if 0 <= slot < slots_per_line: + strong_slots.append(slot) + + def next_strong_slot(position: int) -> int | None: + for slot in strong_slots: + if slot >= position: + return slot + return None + + # Build groups: stressed token plus following unstressed tokens stays together. + groups = [] + idx = 0 + while idx < len(tokens): + token = tokens[idx] + if is_stressed_token(token): + group = [token] + idx += 1 + while idx < len(tokens) and not is_stressed_token(tokens[idx]): + group.append(tokens[idx]) + idx += 1 + groups.append((True, group)) + else: + group = [token] + idx += 1 + while idx < len(tokens) and not is_stressed_token(tokens[idx]): + group.append(tokens[idx]) + idx += 1 + groups.append((False, group)) + + lines = [] + line_tokens = [] + group_i = 0 + + if lead_in_beats is not None: + forced_lead = max(0, lead_in_beats) + line_tokens.extend(["~"] * min(forced_lead, slots_per_line)) + + while group_i < len(groups): + is_stress_group, group = groups[group_i] + cursor = len(line_tokens) + + if is_stress_group: + target_slot = next_strong_slot(cursor) + if target_slot is None: + line_tokens.extend(["~"] * (slots_per_line - len(line_tokens))) + lines.append(" ".join(line_tokens)) + line_tokens = [] + continue + + needed_rests = max(0, target_slot - cursor) + if cursor + needed_rests + len(group) > slots_per_line: + line_tokens.extend(["~"] * (slots_per_line - len(line_tokens))) + lines.append(" ".join(line_tokens)) + line_tokens = [] + continue + + line_tokens.extend(["~"] * needed_rests) + line_tokens.extend(group) + group_i += 1 + continue + + # Unstressed/OOV group: pack directly, split only if needed. + available = slots_per_line - cursor + if len(group) <= available: + line_tokens.extend(group) + group_i += 1 + else: + if available > 0: + line_tokens.extend(group[:available]) + groups[group_i] = (False, group[available:]) + line_tokens.extend(["~"] * (slots_per_line - len(line_tokens))) + lines.append(" ".join(line_tokens)) + line_tokens = [] + + if len(line_tokens) == slots_per_line: + lines.append(" ".join(line_tokens)) + line_tokens = [] + + if line_tokens: + line_tokens.extend(["~"] * (slots_per_line - len(line_tokens))) + lines.append(" ".join(line_tokens)) + + return lines + + +def normalize_phoneme_definition(definition: str) -> str: + """Normalize textarea input into the phoneme API definition format.""" + definition = definition.strip().lower() + definition = re.sub(r"\r?\n+", ",", definition) + definition = re.sub(r"\s*,\s*", ",", definition) + definition = re.sub(r"[ \t]+", "_", definition) + definition = re.sub(r"[^a-z0-9_,]+", "", definition) + definition = re.sub(r"_+", "_", definition) + definition = re.sub(r",+", ",", definition) + return definition.strip("_,") + + +def parse_bool_arg(name: str, default: bool = False) -> bool: + """Parse query bools robustly (supports 1/0, true/false, yes/no, on/off).""" + raw = request.args.get(name) + if raw is None: + return default + value = str(raw).strip().lower() + if value in {"1", "true", "t", "yes", "y", "on"}: + return True + if value in {"0", "false", "f", "no", "n", "off", ""}: + return False + return default @bp.route("/") @@ -82,18 +298,17 @@ async def pack(definition): @bp.route("/.json") async def pack_json(definition): """Download a reslist definition""" - complete = request.args.get("complete", False, type=bool) + complete = parse_bool_arg("complete", False) licenses = request.args.get("licenses", None) - strudel = request.args.get("strudel", False, type=bool) + strudel = parse_bool_arg("strudel", False) if licenses is not None: licenses = licenses.split(",") await pack(definition) url = urlparse(request.base_url) - # SORRY :( QUICK HACK TO SUPPORT MY REVERSE PROXY - # base = url.scheme + "://" + url.hostname - base = "https://" + url.hostname + scheme = url.scheme if url.hostname in ("localhost", "127.0.0.1") else "https" + base = scheme + "://" + url.hostname if url.port: base += ":" + str(url.port) base += "/" @@ -182,6 +397,7 @@ async def speech(definition): """Download a spoken word""" gender = request.args.get("gender", "f") language = request.args.get("language", "en-GB") + force = parse_bool_arg("force", False) definition = definition.replace(" ", "_") try: @@ -190,7 +406,7 @@ async def speech(definition): raise BadRequest(ex) from ex tasks = [] for word in words: - tasks.append(speak_one(word, language, gender)) + tasks.append(speak_one(word, language, gender, force=force)) results = await asyncio.gather(*tasks) global_status = "empty" for status in results: @@ -205,10 +421,347 @@ async def speech(definition): ) +async def _generate_phoneme_samples(definition, language, gender, force, overrides): + """Synthesise all samples needed for a definition string.""" + sentence_definitions = [part for part in definition.split(",") if part] + sentence_word_phones = [ + phonemize.sentence_to_arpabet(sentence_def, overrides=overrides) + for sentence_def in sentence_definitions + ] + + tasks = [] + seen_chunks = set() + for sentence in sentence_word_phones: + for unit in sentence_chunk_plan(sentence): + unit_type = unit[0] + if unit_type == "oov": + value = unit[1] + tasks.append( + speak_one(safe_bank_name(value), language, gender, force=force) + ) + continue + + chunk = unit[1] + text_hint = unit[2] + chunk_key_raw = unit[3] + chunk_key = safe_bank_name(chunk_key_raw, preserve_underscores=True) + if chunk_key not in seen_chunks: + seen_chunks.add(chunk_key) + chunk_ipa = [phonemize.arpabet_to_ipa(phone) for phone in chunk] + tasks.append( + speak_phoneme_chunk_one( + chunk_key, + chunk_ipa, + language, + gender, + text_hint=text_hint, + force=force, + ) + ) + + await asyncio.gather(*tasks) + return sentence_word_phones + + +async def build_phoneme_sentence_data(definition, overrides): + """Build phoneme metadata without synthesising audio.""" + sentence_definitions = [part for part in definition.split(",") if part] + sentence_word_phones = [ + phonemize.sentence_to_arpabet(sentence_def, overrides=overrides) + for sentence_def in sentence_definitions + ] + + return sentence_word_phones + + +def collect_phoneme_banks(sentence_word_phones): + """Collect unique TTS bank keys needed for a phoneme definition.""" + banks = [] + seen = set() + + for sentence in sentence_word_phones: + for word, phones in sentence: + if phones is None: + bank = safe_bank_name(word) + if bank and bank not in seen: + seen.add(bank) + banks.append(bank) + + for unit in sentence_chunk_plan(sentence): + if unit[0] != "chunk": + continue + chunk_key_raw = unit[3] + bank = safe_bank_name(chunk_key_raw, preserve_underscores=True) + if bank and bank not in seen: + seen.add(bank) + banks.append(bank) + + return banks + + +def speech_sample_zip_path(sample_file): + """Return a stable relative zip path for speech sample files.""" + marker = SPEECH_SAMPLE_PATH + if marker in sample_file: + return sample_file.split(marker, 1)[1] + return os.path.basename(sample_file) + + +@bp.route("/phonemes/") +async def phonemes(definition): + """Generate phoneme samples for a sentence""" + gender = request.args.get("gender", "f") + language = request.args.get("language", "en-GB") + force = parse_bool_arg("force", False) + overrides = parse_arpabet_overrides(request.args.get("overrides")) + + definition = normalize_phoneme_definition(definition) + sentence_word_phones = await _generate_phoneme_samples( + definition, language, gender, force, overrides + ) + word_phones = [item for sentence in sentence_word_phones for item in sentence] + + global_status = "empty" + words_out = [] + phonemes_out = [] + for word, phones in word_phones: + if phones is None: + words_out.append(word) + phonemes_out.append([word]) # single-element list signals OOV + else: + words_out.append(word) + phonemes_out.append(phones) + global_status = "ok" + if phones: + global_status = "ok" + + return jsonify( + { + "status": global_status, + "words": words_out, + "phonemes": phonemes_out, + } + ) + + +@bp.route("/phonemes/.zip") +async def phonemes_zip(definition): + """Download a zip archive for generated phoneme/OOV speech samples.""" + gender = request.args.get("gender", "f") + language = request.args.get("language", "en-GB") + force = parse_bool_arg("force", False) + overrides = parse_arpabet_overrides(request.args.get("overrides")) + + definition = normalize_phoneme_definition(definition) + sentence_word_phones = await _generate_phoneme_samples( + definition, language, gender, force, overrides + ) + + banks = collect_phoneme_banks(sentence_word_phones) + + with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp_handle: + tmpfile = tmp_handle.name + + with ZipFile(tmpfile, "w") as zipfile: + added = set() + for bank in banks: + samples = dj.list(bank, gender=gender, language=language, soundtype="tts") + for sample in samples: + if not sample.file or not os.path.exists(sample.file): + continue + arcname = speech_sample_zip_path(sample.file) + if arcname in added: + break + zipfile.write(sample.file, arcname) + added.add(arcname) + break + + @after_this_request + def remove_file(response): + os.remove(tmpfile) + return response + + return send_file(tmpfile, as_attachment=True) + + +@bp.route("/phonemes/.json") +async def phonemes_json(definition): + """Return ordered phoneme sequence for reconstructing the sentence""" + strudel = parse_bool_arg("strudel", False) + details = parse_bool_arg("details", False) + gender = request.args.get("gender", "f") + language = request.args.get("language", "en-GB") + beats_per_bar = request.args.get("beats_per_bar", 4, type=int) + bars_per_line = request.args.get("bars_per_line", 2, type=int) + lead_in_beats = request.args.get("lead_in_beats", type=int) + target_stress_beat = request.args.get("target_stress_beat", 3, type=int) + preview = parse_bool_arg("preview", False) + overrides_raw = request.args.get("overrides") + overrides = parse_arpabet_overrides(overrides_raw) + force = parse_bool_arg("force", False) + definition = normalize_phoneme_definition(definition) + + response_cache_key = ( + definition, + strudel, + details, + preview, + gender, + language, + beats_per_bar, + bars_per_line, + lead_in_beats, + target_stress_beat, + json.dumps(overrides, sort_keys=True), + ) + + if not force: + cached_response = _phoneme_response_cache.get(response_cache_key) + if cached_response is not None: + return jsonify(cached_response) + + if preview: + sentence_word_phones = await build_phoneme_sentence_data(definition, overrides) + else: + # Ensure all samples exist (with correct overrides and force flag) + sentence_word_phones = await _generate_phoneme_samples( + definition, language, gender, force, overrides + ) + + url = urlparse(request.base_url) + scheme = url.scheme if url.hostname in ("localhost", "127.0.0.1") else "https" + base = scheme + "://" + url.hostname + if url.port: + base += ":" + str(url.port) + base += "/" + word_phones = [item for sentence in sentence_word_phones for item in sentence] + + sentences_phonetic = [] + for sentence in sentence_word_phones: + sentence_words = [] + for word, phones in sentence: + if phones: + arpabet_string = " ".join(phones) + sentence_words.append( + { + "word": word, + "arpabet": arpabet_string, + "ipa": "".join(phonemize.arpabet_to_ipa(phone) for phone in phones), + "stress_pattern": pronouncing.stresses(arpabet_string), + "stressed_arpabet": [ + phone for phone in phones if phone[-1:].isdigit() and phone[-1] in ("1", "2") + ], + } + ) + else: + sentence_words.append( + { + "word": word, + "arpabet": None, + "ipa": None, + "stress_pattern": None, + "stressed_arpabet": [], + } + ) + sentences_phonetic.append(sentence_words) + + sentences_strudel = [] + sentences_strudel_timed = [] + sentence_chunks = [] + for sentence in sentence_word_phones: + sentence_tokens = [] + for unit in sentence_chunk_plan(sentence): + unit_type = unit[0] + if unit_type == "oov": + value = unit[1] + sentence_tokens.append(safe_bank_name(value)) + else: + chunk_key_raw = unit[3] + chunk_key = safe_bank_name(chunk_key_raw, preserve_underscores=True) + sentence_tokens.append(chunk_key) + if not preview: + sentence_chunks.append(chunk_key) + sentences_strudel.append(" ".join(sentence_tokens)) + sentences_strudel_timed.append( + sentence_tokens_to_timed_lines( + sentence_tokens, + beats_per_bar=beats_per_bar, + bars_per_line=bars_per_line, + lead_in_beats=lead_in_beats, + target_stress_beat=target_stress_beat, + ) + ) + + if strudel: + reslist = {"_base": base} + else: + reslist = [] + + if not preview: + for word, phones in word_phones: + if phones is None: + # OOV: use whole-word sample + bank = safe_bank_name(word) + samples = dj.list(bank, gender=gender, language=language, soundtype="tts") + for sound in samples: + if strudel: + if bank not in reslist: + reslist[bank] = [] + reslist[bank].append(sound.file) + else: + reslist.append( + { + "url": sound.file, + "type": "audio", + "bank": bank, + "n": 0, + "oov": True, + } + ) + break # only one WAV per word/lang/gender + + seen_chunk_banks = set() + for bank in sentence_chunks: + if bank in seen_chunk_banks: + continue + seen_chunk_banks.add(bank) + samples = dj.list(bank, gender=gender, language=language, soundtype="tts") + for sound in samples: + if strudel: + if bank not in reslist: + reslist[bank] = [] + reslist[bank].append(sound.file) + else: + reslist.append( + { + "url": sound.file, + "type": "audio", + "bank": bank, + "n": 0, + "oov": False, + } + ) + break # only one WAV per chunk/lang/gender + + if strudel and details: + reslist["sentences_strudel"] = sentences_strudel + reslist["sentences_strudel_timed"] = sentences_strudel_timed + reslist["sentences_phonetic"] = sentences_phonetic + reslist["beats_per_bar"] = beats_per_bar + reslist["bars_per_line"] = bars_per_line + reslist["lead_in_beats"] = lead_in_beats + reslist["target_stress_beat"] = target_stress_beat + + if not force: + _phoneme_response_cache[response_cache_key] = reslist + + return jsonify(reslist) + + @bp.route("/speech/.json") async def speech_json(definition): """Download a reslist definition""" - strudel = request.args.get("strudel", False, type=bool) + strudel = parse_bool_arg("strudel", False) gender = request.args.get("gender", "f") language = request.args.get("language", "en-GB") definition = definition.replace(" ", "_") @@ -216,8 +769,8 @@ async def speech_json(definition): await speech(definition) url = urlparse(request.base_url) - # base = url.scheme + "://" + url.hostname - base = "https://" + url.hostname + scheme = url.scheme if url.hostname in ("localhost", "127.0.0.1") else "https" + base = scheme + "://" + url.hostname if url.port: base += ":" + str(url.port) base += "/" @@ -309,9 +862,28 @@ def cors_after(response): return response -async def speak_one(word, language, gender): +async def speak_one(word, language, gender, force=False): """Speak a word""" - return await dj.speak(word, language, gender) + return await dj.speak(word, language, gender, force=force) + + +async def speak_phoneme_one(arpabet_phone, ipa, language, gender, force=False): + """Synthesise a single phoneme""" + return await dj.speak_phoneme(arpabet_phone, ipa, language, gender, force=force) + + +async def speak_phoneme_chunk_one( + chunk_key, + ipa_phones, + language, + gender, + text_hint="", + force=False, +): + """Synthesise a chunk of phonemes""" + return await dj.speak_phoneme_chunk( + chunk_key, ipa_phones, language, gender, text_hint=text_hint, force=force + ) async def fetch_one(word, number, licenses): diff --git a/tests/test_web_chunking.py b/tests/test_web_chunking.py new file mode 100644 index 0000000..05d6a7d --- /dev/null +++ b/tests/test_web_chunking.py @@ -0,0 +1,94 @@ +"""Tests for sentence chunk planning.""" + +from __future__ import annotations + +import importlib.util +import sys +import types +from pathlib import Path + + +def _load_web_module(): + package_root = Path(__file__).resolve().parents[1] + + flask = types.ModuleType("flask") + + class Blueprint: + def __init__(self, *args, **kwargs): + pass + + def route(self, *args, **kwargs): + def decorator(func): + return func + + return decorator + + flask.Blueprint = Blueprint + flask.jsonify = lambda *args, **kwargs: None + flask.send_from_directory = lambda *args, **kwargs: None + flask.request = types.SimpleNamespace(args={}) + flask.render_template = lambda *args, **kwargs: "" + flask.send_file = lambda *args, **kwargs: None + flask.after_this_request = lambda func: func + sys.modules["flask"] = flask + + werkzeug = types.ModuleType("werkzeug") + exceptions = types.ModuleType("werkzeug.exceptions") + + class BadRequest(Exception): + pass + + class HTTPException(Exception): + pass + + exceptions.BadRequest = BadRequest + exceptions.HTTPException = HTTPException + werkzeug.exceptions = exceptions + sys.modules["werkzeug"] = werkzeug + sys.modules["werkzeug.exceptions"] = exceptions + + shabda_pkg = types.ModuleType("shabda") + shabda_pkg.__path__ = [str(package_root / "shabda")] + sys.modules["shabda"] = shabda_pkg + + dj_module = types.ModuleType("shabda.dj") + + class Dj: + def __init__(self, *args, **kwargs): + pass + + dj_module.Dj = Dj + sys.modules["shabda.dj"] = dj_module + + client_module = types.ModuleType("shabda.client") + + class FreesoundUnavailableError(Exception): + pass + + client_module.FreesoundUnavailableError = FreesoundUnavailableError + sys.modules["shabda.client"] = client_module + + web_path = package_root / "shabda" / "web.py" + spec = importlib.util.spec_from_file_location("shabda.web", web_path) + module = importlib.util.module_from_spec(spec) + sys.modules["shabda.web"] = module + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def test_sentence_chunk_plan_keeps_each_word_isolated(): + web = _load_web_module() + + plan = web.sentence_chunk_plan( + [ + ("hello", ["HH", "AH0", "L", "OW1"]), + ("a", ["AH0"]), + ] + ) + + assert plan == [ + ("chunk", ["HH", "AH0"], "hello", "phc_HH_AH0"), + ("chunk", ["L", "OW1"], "hello", "phc_L_OW1"), + ("chunk", ["AH0"], "a", "phc_AH0"), + ] \ No newline at end of file