o %ݫi+B@sdZddlZddlZddlmZddlmZddlmZm Z m Z m Z m Z ddl Z ddlmZddlmZdd lmZmZmZeeZe e ee efZd e eefd e eefd dfd dZ d-ded e efddZded e efddZded e eeffddZde ed e eeffddZde e ededed e e efddZ  d.ded!e eefd"e eefd#ed$e!d%e"d ej#fd&d'Z$ d/ded!e eefd"e eefd%e"d ej#f d(d)Z%d0d+d,Z&dS)1ao This module contains functions to prepare the lexicon and the language model for k2 training. It is based on the script `prepare_lang.sh` from k2/icefall (work of Fangjun Kuang). The original script is under Apache 2.0 license. This script is modified to work with SpeechBrain. Modified by: * Pierre Champion 2023 * Zeyu Zhao 2023 * Georgios Karakasidis 2023 N) defaultdict)Path)AnyDictListTupleUnion) get_logger)k2)EPS read_lexicon write_lexiconfilenamesym2idreturncCs\t|ddd}|D]\}}||d|dq WddS1s'wYdS)a9 Write a symbol to ID mapping to a file. NOTE: No need to implement `read_mapping` as it can be done through :func:`k2.SymbolTable.from_file`. Arguments --------- filename: str Filename to save the mapping. sym2id: Dict[str, int] A dict mapping symbols to IDs. wzutf-8)encoding  N)openitemswrite)rrfsymir[/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/k2_integration/prepare_lang.py write_mapping s "rSILFlexiconcCsVt}|r |||D]\}}||vsJ|d|||q tt|}|S)a0 Get tokens from a lexicon. Arguments --------- lexicon: Lexicon It is the return value of :func:`read_lexicon`. sil_token: str The optional silence token between words. It should not appear in the lexicon, otherwise it will cause an error. manually_add_sil_to_tokens: bool If true, add `sil_token` to the tokens. This is useful when the lexicon does not contain `sil_token` but it is needed in the tokens. Returns ------- sorted_ans: List[str] A list of unique tokens. z5 should not appear in the lexicon but it is found in )setaddupdatesortedlist)r sil_tokenmanually_add_sil_to_tokensans_tokens sorted_ansrrr get_tokens3s      r,cCs.t}|D] \}}||qtt|}|S)z Get words from a lexicon. Arguments --------- lexicon: Lexicon It is the return value of :func:`read_lexicon`. Returns ------- sorted_ans: Return a list of unique words. )r!r"r$r%)r r(wordr)r+rrr get_wordsUs    r.c Cs.tt}|D]\}}|d|d7<qtt}|D]\}}|}||r7d|d|<||s*qg}d}|d}tt}|D]L\} }d|} | dksUJ|| dkri|| dkri|| |fqF|| } | dkrt|} n| d7} | |kr~| }| || <| d| 7} || | fqF||fS)a  It adds pseudo-token disambiguation symbols #1, #2 and so on at the ends of tokens to ensure that all pronunciations are different, and that none is a prefix of another. See also add_lex_disambig.pl from kaldi. Arguments --------- lexicon: Lexicon It is returned by :func:`read_lexicon`. Returns ------- ans: The output lexicon with disambiguation symbols max_disambig: The ID of the max disambiguation symbol that appears in the lexicon rr rz #)rintjoincopypopappendsplit) r countr)r*issubseqr(first_allowed_disambig max_disambiglast_used_disambig_symbol_ofr-tokenseq cur_disambigrrradd_disambig_symbolsjs@      r=symbolscCsddt|DS)z Generate ID maps, i.e., map a symbol to a unique ID. Arguments --------- symbols: List[str] A list of unique symbols. Returns ------- A dict containing the mapping between symbols and IDs. cSsi|]\}}||qSrr).0rrrrr z#generate_id_map..) enumerate)r>rrrgenerate_id_maps rCarcsdisambig_token disambig_wordc CsZt}|D]}|\}}}}} |dkr||qg} |D] } | | | ||dgq|| S)a Adds self-loops to states of an FST to propagate disambiguation symbols through it. They are added on each state with non-epsilon output symbols on at least one arc out of the state. See also fstaddselfloops.pl from Kaldi. One difference is that Kaldi uses OpenFst style FSTs and it has multiple final states. This function uses k2 style FSTs and it does not need to add self-loops to the final state. The input label of a self-loop is `disambig_token`, while the output label is `disambig_word`. Arguments --------- arcs: List[List[Any]] A list-of-list. The sublist contains `[src_state, dest_state, label, aux_label, score]` disambig_token: int It is the token ID of the symbol `#0`. disambig_word: int It is the word ID of the symbol `#0`. Returns ------- Return new `arcs` containing self-loops. r)r!r"r4) rDrErFstates_needs_self_loopsarcsrcdstilabelolabelscorer(srrradd_self_loopss rO?token2idword2idr&sil_probneed_self_loopsc s|dkr|dks Jt|}td|}d}d} d} d} g} tdks(J|tdks0Jd} |}| || | | |g| || | | |g| | | || dg|D]k\}}t|dksgJ|d| }||}fdd |D}tt|dD]}|dkr|n| }| || |||dg| }| d7} q~t|d}|dkr|n| }| || ||||g| || ||||gqV|rӈd }|d }t| ||d } | }| | |d d dg| |gt| d dd} dd | D} dd | D} d| } t j j | dd}|S)aR Convert a lexicon to an FST (in k2 format) with optional silence at the beginning and end of each word. Arguments --------- lexicon: Lexicon The input lexicon. See also :func:`read_lexicon` token2id: Dict[str, int] A dict mapping tokens to IDs. word2id: Dict[str, int] A dict mapping words to IDs. sil_token: str The silence token. sil_prob: float The probability for adding a silence at the beginning and end of the word. need_self_loops: bool If True, add self-loop to states with non-epsilon output symbols on at least one arc out of the state. The input label for this self loop is `token2id["#0"]` and the output label is `word2id["#0"]`. Returns ------- fsa: k2.Fsa An FSA representing the given lexicon. gg?rr  has no pronunciationscg|]}|qSrrr?rrQrr +z"lexicon_to_fst..#0rErFcS|dSNrrrHrrrIz lexicon_to_fst..keycSg|] }dd|DqS)cSg|]}t|qSrstrrYrrrr[Jr\z-lexicon_to_fst...rr?rHrrrr[JcSg|]}d|qSrr1rkrrrr[KrArFacceptor) mathlogr r4lenrangerOr$r1r Fsafrom_str)r rQrRr&rSrT sil_score no_sil_score start_state loop_state sil_state next_staterDeps sil_token_idr-r* cur_staterrrErF final_statefsarrZrlexicon_to_fsts\#      rc sd}d}g}tdksJ|tdksJd}|D]_\}} t| dks+J|d|} ||}fdd| D} tt| dD]} | dkrJ|n|} || || | | dg|} |d7}qBt| d} | dkrk|n|} || || | | dgq|rd} |d}t|| |d}|}|||dddg||gt|d d d }d d|D}d d|D}d|}tjj |dd}|S)au Convert a lexicon to an FST (in k2 format). Arguments --------- lexicon: Lexicon The input lexicon. See also :func:`read_lexicon` token2id: Dict[str, int] A dict mapping tokens to IDs. word2id: Dict[str, int] A dict mapping words to IDs. need_self_loops: bool If True, add self-loop to states with non-epsilon output symbols on at least one arc out of the state. The input label for this self loop is `token2id["#0"]` and the output label is `word2id["#0"]`. Returns ------- fsa: k2.Fsa An FSA representing the given lexicon. rr rWcrXrrrYrZrrr[|r\z)lexicon_to_fst_no_sil..r]r^r_cSr`rarrbrrrrcrdz'lexicon_to_fst_no_sil..recSrg)cSrhrrirYrrrr[r\z4lexicon_to_fst_no_sil...rrkrrrr[rlcSrmrnrorkrrrr[rArFrp) r rtrur4rOr$r1r rvrw)r rQrRrTr{r}rDr~r-piecesrrrrErFrrrrZrlexicon_to_fst_no_silRsH     rTc Cst|}|d}|r'|dr'|dj|jkr'td|ddSdD]-}||rVtj|ddd td ||d |d |t |||d|q)t t |}|d krit ||dd}nt |dd}t |} t|\} } t| dD]} d| } | |vsJ|d| qt|vsJtg|}t| vsJd| vsJd| vsJd| vsJtg| gd} t|}t| }td|dt|d|t|d|t|d| |d krt|||||d}nt|||d}|d kr t| ||||dd}nt| ||dd}t|}td |dt||d!t||d"t||ddS)#aE This function takes as input a lexicon file "$lang_dir/lexicon.txt" consisting of words and tokens (i.e., phones) and does the following: 1. Add disambiguation symbols to the lexicon and generate lexicon_disambig.txt 2. Generate tokens.txt, the token table mapping a token to a unique integer. 3. Generate words.txt, the word table mapping a word to a unique integer. 4. Generate L.pt, in k2 format. It can be loaded by d = torch.load("L.pt") lexicon = k2.Fsa.from_dict(d) 5. Generate L_disambig.pt, in k2 format. Arguments --------- lang_dir: str The directory to store the output files and read the input file lexicon.txt. sil_token: str The silence token. Default is "SIL". sil_prob: float The probability for adding a silence at the beginning and end of the word. Default is 0.5. cache: bool Whether or not to load/cache from/to the .pt format. Returns ------- None Example ------- >>> from speechbrain.k2_integration.prepare_lang import prepare_lang >>> # Create a small lexicon containing only two words and write it to a file. >>> lang_tmpdir = getfixture('tmpdir') >>> lexicon_sample = '''hello h e l l o\nworld w o r l d''' >>> lexicon_file = lang_tmpdir.join("lexicon.txt") >>> lexicon_file.write(lexicon_sample) >>> prepare_lang(lang_tmpdir) >>> for expected_file in ["tokens.txt", "words.txt", "L.pt", "L_disambig.pt", "Linv.pt" ]: ... assert os.path.exists(os.path.join(lang_tmpdir, expected_file)) z lexicon.txtLinv.ptzSkipping lang preparation of 'zA'. Set 'caching: False' in the yaml if this is not what you want.N)L.pt L_disambig.pt tokens.txt words.txtrlexicon_disambig.txtbackupT)exist_okz Backing up z to z/backup/r)r&r'F)r'r #r])r]rrz7Saving tokens.txt, words.txt, lexicon_disambig.txt to ''rrr)rQrRr&rS)rQrR)rQrRr&rSrT)rQrRrTz(Saving L.pt, Linv.pt, L_disambig.pt to 'rr)rexistsstatst_mtimeloggerwarningosmakedirsdebugrenamer rjr,r.r=rur4r rCinforrrrr arc_sortinverttorchsaveas_dict)lang_dirr&rScacheout_dirlexicon_filenamerr r*wordslexicon_disambigr9rdisambigrQrRL L_disambigL_invrrr prepare_langs2                   r)rF)rrPF)F)rrPT)'__doc__rrr collectionsrpathlibrtypingrrrrrrspeechbrain.utils.loggerr r/r r r r r__name__rrjLexiconr0rr,r.r=rCrOfloatboolrvrrrrrrrs|     & "G   /   j   N