FfmpegConvert

Signed-off-by: Nikolay Karpov <[email protected]>

dataset_configs/commoncrawl/small_sentence.yaml

@@ -2,33 +2,29 @@ processors_to_run: "0:"
 workspace_dir: /mnt/ssd8/cc_sdp
 
 processors:
-  - _target_: sdp.processors.datasets.commoncrawl.CreateInitialManifestCC
-    raw_data_dir: /mnt/ssd8/cc_download/videos_and_transcripts_large_16_cores_output/video_output2
+  - _target_: sdp.processors.datasets.commoncrawl.FfmpegConvert
+    input_manifest_file: ${workspace_dir}/manifest_urls.json
     output_manifest_file: ${workspace_dir}/manifest0.json
     resampled_audio_dir: ${workspace_dir}/audio/
     target_samplerate: 16000
     target_nchannels: 1
-    audio_field: "audios"
     video_field: "videos"
+    audio_field: "audios"
     key_field: "key"
-    text_field: "texts"
 
   - _target_: sdp.processors.datasets.commoncrawl.TxtToVtt
-    input_manifest_file: ${workspace_dir}/manifest0.json
     output_manifest_file: ${workspace_dir}/manifest1.json
     vtt_files_dir: ${workspace_dir}/vtts/
     key_field: "key"
     text_field: "texts"
     vtt_field: "vtt_filepath"
 
-  - _target_: sdp.processors.datasets.commoncrawl.AllVttText 
-    input_manifest_file: ${workspace_dir}/manifest1.json
+  - _target_: sdp.processors.datasets.commoncrawl.AllVttText
     output_manifest_file: ${workspace_dir}/manifest2.json
     input_filepath_field: vtt_filepath
     output_text_field: vtt_text
 
   - _target_: sdp.processors.datasets.commoncrawl.TextLid
-    input_manifest_file: ${workspace_dir}/manifest2.json
     output_manifest_file: ${workspace_dir}/manifest3.json
     input_text_field: vtt_text
     output_lang_field: text_lang
@@ -47,25 +43,22 @@ processors:
     device: cuda
     pretrained_model: "langid_ambernet"
 
-  - _target_: sdp.processors.datasets.commoncrawl.ReadParquet
-    raw_data_dir: /mnt/ssd8/cc_download/videos_and_transcripts_large_16_cores_output/video_output2
-    output_manifest_file: ${workspace_dir}/manifest6.json
-    output_video_field: video
-    output_vtt_field: caption
-    key_field: key
-
   - _target_: sdp.processors.datasets.commoncrawl.SplitByVttSentence
-    output_manifest_file: ${workspace_dir}/manifest7.json
+    output_manifest_file: ${workspace_dir}/manifest6a.json
     splited_audio_dir: ${workspace_dir}/splited_s/
     source_audio_field: audios
     vtt_field: "vtt_filepath"
     target_audio_field: "audio_filepath"
     duration_field: "duration"
     text_field: "text"
-    proxy_fields: [audio_lang, text_lang, video, caption]
+    proxy_fields: [audio_lang, text_lang]
     duration_threshold: 10.0
 
   - _target_: sdp.processors.DropHighLowDuration
-    output_manifest_file: ${workspace_dir}/manifest8.json
+    output_manifest_file: ${workspace_dir}/manifest7a.json
     high_duration_threshold: 40
     low_duration_threshold: 0.02
+
+  - _target_: sdp.processors.KeepOnlySpecifiedFields
+    output_manifest_file: ${workspace_dir}/manifest8a.json
+    fields_to_keep: ["audio_filepath", "duration", "text", "audio_lang", "text_lang"]

sdp/processors/datasets/commoncrawl/__init__.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .commoncrawl import UseSonar, BLEUScore, Subprocess, NmtSubprocess, PreserveByValue, Lang2Iso, SplitByVttSentence, SplitByVtt, AudioLid, TextLid, AllVttText, TxtToVtt, ReadParquet, CreateInitialManifestCC
+from .commoncrawl import UseSonar, BLEUScore, Subprocess, NmtSubprocess, PreserveByValue, Lang2Iso, SplitByVttSentence, SplitByVtt, AudioLid, TextLid, AllVttText, TxtToVtt, ReadParquet, CreateInitialManifestCC, FfmpegConvert

sdp/processors/datasets/commoncrawl/commoncrawl.py

@@ -339,8 +339,9 @@ def process_dataset_entry(self, data_entry):
     def makeDataEntry(self, data_entry, data, vtt_file, samplerate, text_c, start_c, end_c):
         data_sample = data[start_c:end_c]
         wav_save_file = os.path.join(self.splited_audio_dir, '/'.join(os.path.splitext(vtt_file)[0].split('/')[-2:]), str(int(start_c/(samplerate/1000)))+"-"+str(int(end_c/(samplerate/1000)))+".wav")
-        os.makedirs(os.path.split(wav_save_file)[0], exist_ok=True)
-        sf.write(wav_save_file, data_sample, samplerate)
+        if not os.path.isfile(wav_save_file):
+            os.makedirs(os.path.split(wav_save_file)[0], exist_ok=True)
+            sf.write(wav_save_file, data_sample, samplerate)
 
         data = {self.target_audio_field: wav_save_file,
                     self.duration_field: data_sample.shape[0]/samplerate,
@@ -663,9 +664,52 @@ def process_dataset_entry(self, data_entry):
         (video,	key, text) = data_entry
         os.makedirs(os.path.join(self.resampled_audio_dir, key.split("/")[0]), exist_ok=True)
         audio = os.path.join(self.resampled_audio_dir, key) + ".wav"
-        ffmpeg_convert(video, audio, self.target_samplerate, self.target_nchannels)
+        if not os.path.isfile(audio):
+            ffmpeg_convert(video, audio, self.target_samplerate, self.target_nchannels)
 
         data = {self.audio_field: audio,
+                self.video_field: video,
                 self.key_field: key,
                 self.text_field: text}
         return [DataEntry(data=data)]
+
+class FfmpegConvert(BaseParallelProcessor):
+    """
+        Args:
+        video_field (str): field with path to video file in the input manifest
+        audio_field (str): field with path to audio file in the output manifest
+        key_field (str): field with key value
+        resampled_audio_dir (str): where to put re-sampled and trimmed wav files.
+        target_samplerate (int): sample rate to resample to. Defaults to 16000.
+        target_nchannels (int): target number of channels. Defaults to 1.
+    """
+    def __init__(
+        self,
+        resampled_audio_dir: str,
+        video_field: str,
+        audio_field: str,
+        key_field: str,
+        target_samplerate: int = 16000,
+        target_nchannels: int = 1,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.audio_field = audio_field
+        self.video_field = video_field
+        self.key_field = key_field
+        self.resampled_audio_dir = resampled_audio_dir
+        self.target_samplerate = target_samplerate
+        self.target_nchannels = target_nchannels
+
+    def process_dataset_entry(self, data_entry):
+        video = data_entry[self.video_field]
+        key = os.path.splitext(data_entry[self.video_field])[0][-13:]
+        os.makedirs(os.path.join(self.resampled_audio_dir, key.split("/")[0]), exist_ok=True)
+        audio = os.path.join(self.resampled_audio_dir, key) + ".wav"
+
+        if not os.path.isfile(audio):
+            ffmpeg_convert(video, audio, self.target_samplerate, self.target_nchannels)
+
+        data_entry[self.audio_field]= audio
+        data_entry[self.key_field] = key
+        return [DataEntry(data=data_entry)]