montreal_forced_aligner/corpus/acoustic_corpus.py
def _load_corpus_from_source(self) -> None:
        """
        Load a corpus without using multiprocessing
        """
        begin_time = time.time()
        sanitize_function = None
        if hasattr(self, "sanitize_function"):
            sanitize_function = self.sanitize_function

        all_sound_files = {}
        use_audio_directory = False
        if self.audio_directory and os.path.exists(self.audio_directory):
            use_audio_directory = True
            for root, _, files in os.walk(self.audio_directory, followlinks=True):
                if self.stopped.stop_check():
                    return
                exts = find_exts(files)
                exts.wav_files = {k: os.path.join(root, v) for k, v in exts.wav_files.items()}
                exts.other_audio_files = {
                    k: os.path.join(root, v) for k, v in exts.other_audio_files.items()
                }
                all_sound_files.update(exts.other_audio_files)
                all_sound_files.update(exts.wav_files)
        self.log_debug(f"Walking through {self.corpus_directory}...")
        for root, _, files in os.walk(self.corpus_directory, followlinks=True):
            exts = find_exts(files)
            relative_path = root.replace(self.corpus_directory, "").lstrip("/").lstrip("\\")
            if self.stopped.stop_check():
                return
            self.log_debug(f"Inside relative root {relative_path}:")
            self.log_debug(f"    Found {len(exts.identifiers)} identifiers")
            self.log_debug(f"    Found {len(exts.wav_files)} .wav files")
            self.log_debug(f"    Found {len(exts.other_audio_files)} other audio files")
            self.log_debug(f"    Found {len(exts.lab_files)} .lab files")
            self.log_debug(f"    Found {len(exts.textgrid_files)} .TextGrid files")
            if not use_audio_directory:
                all_sound_files = {}
                wav_files = {k: os.path.join(root, v) for k, v in exts.wav_files.items()}
                other_audio_files = {
                    k: os.path.join(root, v) for k, v in exts.other_audio_files.items()
                }
                all_sound_files.update(other_audio_files)
                all_sound_files.update(wav_files)
            for file_name in exts.identifiers:

                wav_path = None
                transcription_path = None
                if file_name in all_sound_files:
                    wav_path = all_sound_files[file_name]
                if file_name in exts.lab_files:
                    lab_name = exts.lab_files[file_name]
                    transcription_path = os.path.join(root, lab_name)
                elif file_name in exts.textgrid_files:
                    tg_name = exts.textgrid_files[file_name]
                    transcription_path = os.path.join(root, tg_name)
                if wav_path is None and transcription_path is None:  # Not a file for MFA
                    continue
                if wav_path is None:
                    self.transcriptions_without_wavs.append(transcription_path)
                    continue
                if transcription_path is None:
                    self.no_transcription_files.append(wav_path)
                try:
                    file = File.parse_file(
                        file_name,
                        wav_path,
                        transcription_path,
                        relative_path,
                        self.speaker_characters,
                        sanitize_function,
                    )
                    self.add_file(file)
                except TextParseError as e:
                    self.decode_error_files.append(e)
                except TextGridParseError as e:
                    self.textgrid_read_errors[e.file_name] = e
        if self.decode_error_files or self.textgrid_read_errors:
            self.log_info(
                "There were some issues with files in the corpus. "
                "Please look at the log file or run the validator for more information."
            )
            if self.decode_error_files:
                self.log_debug(
                    f"There were {len(self.decode_error_files)} errors decoding text files:"
                )
                self.log_debug(", ".join(self.decode_error_files))
            if self.textgrid_read_errors:
                self.log_debug(
                    f"There were {len(self.textgrid_read_errors)} errors decoding reading TextGrid files:"
                )
                for f, e in self.textgrid_read_errors.items():
                    self.log_debug(f"{f}: {e.error}")

        self.log_debug(f"Parsed corpus directory in {time.time() - begin_time} seconds")

        Load a corpus without using multiprocessing
        
Load a corpus without using multiprocessing
Load a corpus without using multiprocessing
def _load_corpus_from_source(self) -> None:
        begin_time = time.time()
        sanitize_function = None
        if hasattr(self, "sanitize_function"):
            sanitize_function = self.sanitize_function
        all_sound_files = {}
        use_audio_directory = False
        if self.audio_directory and os.path.exists(self.audio_directory):
            use_audio_directory = True
            for root, _, files in os.walk(self.audio_directory, followlinks=True):
                if self.stopped.stop_check():
                    return
                exts = find_exts(files)
                exts.wav_files = {k: os.path.join(root, v) for k, v in exts.wav_files.items()}
                exts.other_audio_files = {
                    k: os.path.join(root, v) for k, v in exts.other_audio_files.items()
                }
                all_sound_files.update(exts.other_audio_files)
                all_sound_files.update(exts.wav_files)
        self.log_debug(f"Walking through {self.corpus_directory}...")
        for root, _, files in os.walk(self.corpus_directory, followlinks=True):
            exts = find_exts(files)
            relative_path = root.replace(self.corpus_directory, "").lstrip("/").lstrip("\\")
            if self.stopped.stop_check():
                return
            self.log_debug(f"Inside relative root {relative_path}:")
            self.log_debug(f"    Found {len(exts.identifiers)} identifiers")
            self.log_debug(f"    Found {len(exts.wav_files)} .wav files")
            self.log_debug(f"    Found {len(exts.other_audio_files)} other audio files")
            self.log_debug(f"    Found {len(exts.lab_files)} .lab files")
            self.log_debug(f"    Found {len(exts.textgrid_files)} .TextGrid files")
            if not use_audio_directory:
                all_sound_files = {}
                wav_files = {k: os.path.join(root, v) for k, v in exts.wav_files.items()}
                other_audio_files = {
                    k: os.path.join(root, v) for k, v in exts.other_audio_files.items()
                }
                all_sound_files.update(other_audio_files)
                all_sound_files.update(wav_files)
            for file_name in exts.identifiers:
                wav_path = None
                transcription_path = None
                if file_name in all_sound_files:
                    wav_path = all_sound_files[file_name]
                if file_name in exts.lab_files:
                    lab_name = exts.lab_files[file_name]
                    transcription_path = os.path.join(root, lab_name)
                elif file_name in exts.textgrid_files:
                    tg_name = exts.textgrid_files[file_name]
                    transcription_path = os.path.join(root, tg_name)
                if wav_path is None and transcription_path is None:  
                    continue
                if wav_path is None:
                    self.transcriptions_without_wavs.append(transcription_path)
                    continue
                if transcription_path is None:
                    self.no_transcription_files.append(wav_path)
                try:
                    file = File.parse_file(
                        file_name,
                        wav_path,
                        transcription_path,
                        relative_path,
                        self.speaker_characters,
                        sanitize_function,
                    )
                    self.add_file(file)
                except TextParseError as e:
                    self.decode_error_files.append(e)
                except TextGridParseError as e:
                    self.textgrid_read_errors[e.file_name] = e
        if self.decode_error_files or self.textgrid_read_errors:
            self.log_info(
                "There were some issues with files in the corpus. "
                "Please look at the log file or run the validator for more information."
            )
            if self.decode_error_files:
                self.log_debug(
                    f"There were {len(self.decode_error_files)} errors decoding text files:"
                )
                self.log_debug(", ".join(self.decode_error_files))
            if self.textgrid_read_errors:
                self.log_debug(
                    f"There were {len(self.textgrid_read_errors)} errors decoding reading TextGrid files:"
                )
                for f, e in self.textgrid_read_errors.items():
                    self.log_debug(f"{f}: {e.error}")
        self.log_debug(f"Parsed corpus directory in {time.time() - begin_time} seconds")
def _load_corpus_from_source ( self ) -> None : begin_time = time . time ( ) sanitize_function = None if hasattr ( self , "sanitize_function" ) : sanitize_function = self . sanitize_function all_sound_files = { } use_audio_directory = False if self . audio_directory and os . path . exists ( self . audio_directory ) : use_audio_directory = True for root , _ , files in os . walk ( self . audio_directory , followlinks = True ) : if self . stopped . stop_check ( ) : return exts = find_exts ( files ) exts . wav_files = { k : os . path . join ( root , v ) for k , v in exts . wav_files . items ( ) } exts . other_audio_files = { k : os . path . join ( root , v ) for k , v in exts . other_audio_files . items ( ) } all_sound_files . update ( exts . other_audio_files ) all_sound_files . update ( exts . wav_files ) self . log_debug ( f"Walking through {self.corpus_directory}..." ) for root , _ , files in os . walk ( self . corpus_directory , followlinks = True ) : exts = find_exts ( files ) relative_path = root . replace ( self . corpus_directory , "" ) . lstrip ( "/" ) . lstrip ( "\\" ) if self . stopped . stop_check ( ) : return self . log_debug ( f"Inside relative root {relative_path}:" ) self . log_debug ( f"    Found {len(exts.identifiers)} identifiers" ) self . log_debug ( f"    Found {len(exts.wav_files)} .wav files" ) self . log_debug ( f"    Found {len(exts.other_audio_files)} other audio files" ) self . log_debug ( f"    Found {len(exts.lab_files)} .lab files" ) self . log_debug ( f"    Found {len(exts.textgrid_files)} .TextGrid files" ) if not use_audio_directory : all_sound_files = { } wav_files = { k : os . path . join ( root , v ) for k , v in exts . wav_files . items ( ) } other_audio_files = { k : os . path . join ( root , v ) for k , v in exts . other_audio_files . items ( ) } all_sound_files . update ( other_audio_files ) all_sound_files . update ( wav_files ) for file_name in exts . identifiers : wav_path = None transcription_path = None if file_name in all_sound_files : wav_path = all_sound_files [ file_name ] if file_name in exts . lab_files : lab_name = exts . lab_files [ file_name ] transcription_path = os . path . join ( root , lab_name ) elif file_name in exts . textgrid_files : tg_name = exts . textgrid_files [ file_name ] transcription_path = os . path . join ( root , tg_name ) if wav_path is None and transcription_path is None : continue if wav_path is None : self . transcriptions_without_wavs . append ( transcription_path ) continue if transcription_path is None : self . no_transcription_files . append ( wav_path ) try : file = File . parse_file ( file_name , wav_path , transcription_path , relative_path , self . speaker_characters , sanitize_function , ) self . add_file ( file ) except TextParseError as e : self . decode_error_files . append ( e ) except TextGridParseError as e : self . textgrid_read_errors [ e . file_name ] = e if self . decode_error_files or self . textgrid_read_errors : self . log_info ( "There were some issues with files in the corpus. " "Please look at the log file or run the validator for more information." ) if self . decode_error_files : self . log_debug ( f"There were {len(self.decode_error_files)} errors decoding text files:" ) self . log_debug ( ", " . join ( self . decode_error_files ) ) if self . textgrid_read_errors : self . log_debug ( f"There were {len(self.textgrid_read_errors)} errors decoding reading TextGrid files:" ) for f , e in self . textgrid_read_errors . items ( ) : self . log_debug ( f"{f}: {e.error}" ) self . log_debug ( f"Parsed corpus directory in {time.time() - begin_time} seconds" )
Load a corpus without using multiprocessing
Load a corpus without using multiprocessing
"""
        Load a corpus without using multiprocessing
        """ # Not a file for MFA
self