initial commit (`f74eed7` in https://bitbucket.org/maum-system/cvpr22-demo-gradio)

.dockerignore

@@ -0,0 +1,2 @@
+output_file/*
+!output_file/.gitkeep

.gitattributes

@@ -0,0 +1,31 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bin.* filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zstandard filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.mp4 filter=lfs diff=lfs merge=lfs -text
+*.png filter=lfs diff=lfs merge=lfs -text
+output_file/* filter=lfs diff=lfs merge=lfs -text
+background_image/* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,13 @@
+.DS_Store
+flagged/
+__pycache__/
+.vscode/
+output_file/*
+
+!output_file/.gitkeep
+
+*.mp4
+*.png
+!background_image/*
+*.mkv
+gradio_queue.db*

README.md

@@ -0,0 +1,46 @@
+---
+title: Talking Face Generation with Multilingual TTS
+emoji: 👄
+colorFrom: blue
+colorTo: blue
+sdk: gradio
+app_file: app.py
+pinned: false
+license: cc-by-nc-sa-4.0
+---
+
+# Configuration
+
+`title`: _string_
+Display title for the Space
+
+`emoji`: _string_
+Space emoji (emoji-only character allowed)
+
+`colorFrom`: _string_
+Color for Thumbnail gradient (red, yellow, green, blue, indigo, purple, pink, gray)
+
+`colorTo`: _string_
+Color for Thumbnail gradient (red, yellow, green, blue, indigo, purple, pink, gray)
+
+`sdk`: _string_
+Can be either `gradio`, `streamlit`, or `static`
+
+`sdk_version` : _string_
+Only applicable for `streamlit` SDK.  
+See [doc](https://hf.co/docs/hub/spaces) for more info on supported versions.
+
+`app_file`: _string_
+Path to your main application file (which contains either `gradio` or `streamlit` Python code, or `static` html code).
+Path is relative to the root of the repository.
+
+`models`: _List[string]_
+HF model IDs (like "gpt2" or "deepset/roberta-base-squad2") used in the Space.
+Will be parsed automatically from your code if not specified here.
+
+`datasets`: _List[string]_
+HF dataset IDs (like "common_voice" or "oscar-corpus/OSCAR-2109") used in the Space.
+Will be parsed automatically from your code if not specified here.
+
+`pinned`: _boolean_
+Whether the Space stays on top of your list.

app.py

@@ -0,0 +1,214 @@
+import os
+import subprocess
+
+REST_IP = os.environ['REST_IP']
+SERVICE_PORT = int(os.environ['SERVICE_PORT'])
+TRANSLATION_APIKEY_URL = os.environ['TRANSLATION_APIKEY_URL']
+GOOGLE_APPLICATION_CREDENTIALS = os.environ['GOOGLE_APPLICATION_CREDENTIALS']
+subprocess.call(f"wget --no-check-certificate -O {GOOGLE_APPLICATION_CREDENTIALS} {TRANSLATION_APIKEY_URL}", shell=True)
+
+import gradio as gr
+from client_rest import RestAPIApplication
+from pathlib import Path
+import argparse
+import threading
+from translator import GoogleAuthTranslation
+import yaml
+
+TITLE = Path("docs/title.txt").read_text()
+DESCRIPTION = Path("docs/description.txt").read_text()
+
+class Translator:
+    def __init__(self, yaml_path='lang.yaml'):
+        self.google_translation = GoogleAuthTranslation(project_id="cvpr-2022-demonstration")
+        with open(yaml_path) as f:
+            self.supporting_languages = yaml.load(f, Loader=yaml.FullLoader)
+            
+    def _get_text_with_lang(self, text, lang):
+        lang_detected = self.google_translation.detect(text)
+        print(lang_detected, lang)
+        if lang is None:
+            lang = lang_detected
+            
+        if lang != lang_detected:
+            target_text = self.google_translation.translate(text, lang=lang)
+        else:
+            target_text = text
+            
+        return target_text, lang
+    
+    def _convert_lang_from_index(self, lang):
+        lang_finder = [name for name in self.supporting_languages
+                        if self.supporting_languages[name]['language'] == lang]
+        if len(lang_finder) == 1:
+            lang = lang_finder[0]
+        else:
+            raise AssertionError(f"Given language index can't be understood! | lang: {lang}")
+        
+        return lang
+
+    def get_translation(self, text, lang, use_translation=True):
+        lang_ = self._convert_lang_from_index(lang)
+        
+        if use_translation:
+            target_text, _ = self._get_text_with_lang(text, lang_)
+        else:
+            target_text = text
+
+        return target_text, lang 
+    
+    
+class GradioApplication:
+    def __init__(self, rest_ip, rest_port, max_seed):
+        self.lang_list = {
+            'Korean': 'ko_KR',
+            'English': 'en_US',
+            'Japanese': 'ja_JP',
+            'Chinese': 'zh_CN'
+        }
+        self.background_list = [None,
+                                "background_image/cvpr.png",
+                                "background_image/black.png",
+                                "background_image/river.mp4",
+                                "background_image/sky.mp4"]
+        
+        self.translator = Translator()
+        self.rest_application = RestAPIApplication(rest_ip, rest_port)
+        self.output_dir = Path("output_file")
+
+        inputs = prepare_input()
+        outputs = prepare_output()
+
+        self.iface = gr.Interface(fn=self.infer,
+                                  title=TITLE,
+                                  description=DESCRIPTION,
+                                  inputs=inputs,
+                                  outputs=outputs,
+                                  allow_flagging='never',
+                                  article=Path("docs/article.md").read_text())
+
+        self.max_seed = max_seed
+        self._file_seed = 0
+        self.lock = threading.Lock()
+        
+    
+    def _get_file_seed(self):
+        return f"{self._file_seed % self.max_seed:02d}"
+
+    def _reset_file_seed(self):
+        self._file_seed = 0
+
+    def _counter_file_seed(self):
+        with self.lock:
+            self._file_seed += 1
+
+    def get_lang_code(self, lang):
+        return self.lang_list[lang]
+
+    def get_background_data(self, background_index):
+        # get background filename and its extension
+        data_path = self.background_list[background_index]
+
+        if data_path is not None:
+            with open(data_path, 'rb') as rf:
+                background_data = rf.read()
+            is_video_background = str(data_path).endswith(".mp4")
+        else:
+            background_data = None
+            is_video_background = False
+
+        return background_data, is_video_background
+
+    def infer(self, text, lang, duration_rate, pad_begin, pad_end, action, background_index):
+        self._counter_file_seed()
+        print(f"File Seed: {self._file_seed}")
+        target_text, lang_dest = self.translator.get_translation(text, lang)
+        lang_rpc_code = self.get_lang_code(lang_dest)
+
+        background_data, is_video_background = self.get_background_data(background_index)
+        
+        video_data = self.rest_application.get_video(target_text, lang_rpc_code, duration_rate, pad_begin, pad_end, action.lower(),
+                                                     background_data, is_video_background)
+        print(len(video_data))
+
+        video_filename = self.output_dir / f"{self._file_seed:02d}.mp4"
+        with open(video_filename, "wb") as video_file:
+            video_file.write(video_data)
+        
+        return f"Language: {lang_dest}\nText: \n{target_text}", str(video_filename)        
+
+    def run(self, server_port=7860, share=False):
+        try:
+            self.iface.launch(height=900,
+                              share=share, server_port=server_port,
+                              enable_queue=True)
+        
+        except KeyboardInterrupt:
+            gr.close_all()
+
+
+def prepare_input():
+    text_input = gr.inputs.Textbox(lines=2,
+                                   placeholder="Type your text with English, Chinese, Korean, and Japanese.",
+                                   default="Hello, this is demonstration for talking face generation "
+                                   "with multilingual text-to-speech.",
+                                   label="Text")
+    lang_input = gr.inputs.Radio(['Korean', 'English', 'Japanese', 'Chinese'],
+                                 type='value',
+                                 default=None,
+                                 label="Language")
+    duration_rate_input = gr.inputs.Slider(minimum=0.8,
+                                           maximum=1.2,
+                                           step=0.01,
+                                           default=1.0,
+                                           label="Duration (The bigger the value, the slower it pronounces)")
+    start_padding_input = gr.inputs.Slider(minimum=0.0,
+                                           maximum=2.0,
+                                           step=0.1,
+                                           default=0.0,
+                                           label="Start padding (s)")
+    end_padding_input = gr.inputs.Slider(minimum=0.0,
+                                         maximum=2.0,
+                                         step=0.1,
+                                         default=0.0,
+                                         label="End padding (s)")
+    action_input = gr.inputs.Radio(['Default', 'Hand', 'BothHand', 'HandDown', 'Sorry'],
+                                    type='value',
+                                    default='Default',
+                                    label="Select an action...")
+    background_input = gr.inputs.Radio(['None', 'CVPR', 'Black', 'River', 'Sky'],
+                                        type='index',
+                                        default='None',
+                                        label="Background image")
+
+    return [text_input, lang_input,
+            duration_rate_input, start_padding_input, end_padding_input,
+            action_input, background_input]
+
+
+def prepare_output():
+    translation_result_otuput = gr.outputs.Textbox(type="str",
+                                                   label="Translation Result")
+
+    video_output = gr.outputs.Video()
+    return [translation_result_otuput, video_output]
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description='GRADIO DEMO for talking face generation submitted to CVPR2022')
+    parser.add_argument('-p', '--port', dest='gradio_port', type=int, default=7860, help="Port for gradio")
+    parser.add_argument('--rest_ip', type=str, default=REST_IP, help="IP for REST API")
+    parser.add_argument('--rest_port', type=int, default=SERVICE_PORT, help="Port for REST API")
+    parser.add_argument('--max_seed', type=int, default=20, help="Max seed for saving video")
+    parser.add_argument('--share', action='store_true', help='get publicly sharable link')
+    args = parser.parse_args()
+    return args    
+
+
+if __name__ == '__main__':
+    args = parse_args()
+    
+    gradio_application = GradioApplication(args.rest_ip, args.rest_port, args.max_seed)
+    gradio_application.run(server_port=args.gradio_port, share=args.share)
+    

background_image/river.mp4

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e8fedd95028adbabf17eb7fcc67bc37d22dc996cb45878a8a52cc95dcfb21cf3
+size 4523353

background_image/sky.mp4

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f12109f770ba152228370f0386d2fdc9892ff79875d3c3296efde200822a3bbc
+size 2222081

client_rest.py

@@ -0,0 +1,73 @@
+import requests
+import json
+import base64
+import argparse
+
+VIDEO_WIDTH = 1080
+VIDEO_HEIGHT = 1920
+
+class RestAPIApplication:
+    def __init__(self, ip, port):
+        
+        if port < 0:
+            self.post_request_addr = f"http://{ip}/register/"
+            self.post_headers = {"Content-Type": "application/json"}
+            self.generate_addr = (lambda id_: f'http://{ip}/generate/{id_}')
+        else:
+            self.post_request_addr = f"http://{ip}:{port}/register/"
+            self.post_headers = {"Content-Type": "application/json"}
+            self.generate_addr = (lambda id_: f'http://{ip}:{port}/generate/{id_}')
+
+    @staticmethod
+    def _get_json_request(text, lang, duration_rate, pad_begin, pad_end, action, background_data=None, is_video_background=False):
+        request_form = dict()
+
+        request_form['text'] = text
+        request_form['speaker'] = 0
+        request_form['width'] = VIDEO_WIDTH
+        request_form['height'] = VIDEO_HEIGHT
+
+        request_form['action'] = action
+        
+        if background_data is not None:
+            background_base64 = base64.b64encode(background_data).decode("UTF-8")
+        else:
+            background_base64 = ""
+        
+        request_form['background'] = background_base64
+        request_form['durationRate'] = 1.0
+        request_form['isVideoBackground'] = is_video_background
+        request_form['lang'] = lang
+
+        request_as_json = json.dumps(request_form)
+        return request_as_json
+
+    @staticmethod
+    def _get_video_id(results):
+        return json.loads(bytes.decode(results.content))['id']
+
+    def get_video(self, text, lang, duration_rate, pad_begin, pad_end, action, background_data=None, is_video_background=False):
+        request_json = self._get_json_request(text, lang, duration_rate, pad_begin, pad_end, action, background_data, is_video_background)
+        
+        # POST request with jsonified request
+        results = requests.post(self.post_request_addr, headers=self.post_headers, data=request_json)
+        
+        # GET video with the given id
+        video_id = self._get_video_id(results)
+        video_results = requests.get(self.generate_addr(video_id))
+        
+        return video_results.content
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description='REST API interface for talking face generation submitted to CVPR2022')
+    parser.add_argument('-i', '--ip', dest='rest_ip', type=str, default="127.0.0.1", help="IP for REST API")
+    parser.add_argument('-p', '--port', dest='rest_port', type=int, default=8080, help="Port for REST API")
+    args = parser.parse_args()
+    return args    
+
+
+if __name__ == '__main__':
+    args = parse_args()
+    rest_api_application = RestAPIApplication(args.rest_ip, args.rest_port)    

docs/article.md

@@ -0,0 +1,6 @@
+[Youtube](https://youtu.be/F6h0s0M4vBI)
+
+
+
+
+Recent studies in talking face generation have focused on building a train-once-use-everywhere model i.e. a model that will generalize from any source speech to any target identity. A number of works have already claimed this functionality and have added that their models will also generalize to any language. However, we show, using languages from different language families, that these models do not translate well when the training language and the testing language are sufficiently different. We reduce the scope of the problem to building a language-robust talking face generation system on seen identities i.e. the target identity is the same as the training identity. In this work, we introduce a talking face generation system that will generalize to different languages. We evaluate the efficacy of our system using a multilingual text-to-speech system. We also discuss the usage of joint text-to-speech system and the talking face generation system as a neural dubber system.

docs/description.txt

@@ -0,0 +1,4 @@
+CVPR 2022 Demo Track (Round 1) #8 | 
+This system generates talking face video with corresponding text. 
+You can input text with one of four languages, which are Chinese, English, Japanese, and Korean. 
+If your text language and target language is different, it translates the sentence into target language with Google Translation API.

docs/title.txt

@@ -0,0 +1 @@
+Talking Face Generation with Multilingual TTS

lang.yaml

@@ -0,0 +1,20 @@
+ko:
+  index: 1
+  language: "Korean"
+  locale: "ko_KR"
+  google_dest: "ko"
+en:
+  index: 2
+  language: "English"
+  locale: "en_US"
+  google_dest: "en"
+ja:
+  index: 3
+  language: "Japanese"
+  locale: "ja_JP"
+  google_dest: "ja"
+zh:
+  index: 4
+  language: "Chinese"
+  locale: "zh_CN"
+  google_dest: "zh-CN"

requirements.txt

@@ -0,0 +1,6 @@
+gradio
+jinja2
+googletrans==4.0.0-rc1
+PyYAML
+opencv-python
+google-cloud-translate

sample_text.yaml

@@ -0,0 +1,8 @@
+ko:
+  - "안녕하세요? 한국어로 말하고 있습니다."
+en:
+  - "Hello. Now I'm speaking in English."
+zh:
+  - "你好? 我在说普通话。"
+ja:
+  - "こんにちは。 今、日本語で話しています。"

translator/__init__.py

@@ -0,0 +1 @@
+from .v3 import GoogleAuthTranslation

translator/v3.py

@@ -0,0 +1,58 @@
+from google.cloud import translate
+import yaml
+
+
+class GoogleAuthTranslation:
+    def __init__(self, project_id, yaml_path='lang.yaml'):
+        self.translator = translate.TranslationServiceClient()
+        self.location = "global"
+        self.parent = f"projects/{project_id}/locations/{self.location}"
+        
+        with open(yaml_path) as f:
+            self.supporting_languages = yaml.load(f, Loader=yaml.FullLoader)
+
+    def _detect(self, query):
+        response = self.translator.detect_language(
+            request={
+                "parent": self.parent,
+                "content": query,
+                "mime_type": "text/plain",  # mime types: text/plain, text/html
+            }
+        )
+
+        for language in response.languages:
+            # First language is the most confident one
+            return language.language_code
+
+    def _get_dest_from_lang(self, lang):
+        try:
+            return self.supporting_languages[lang]['google_dest']
+
+        except KeyError as e:
+            raise e
+        
+    def _get_lang_from_dest(self, dest):
+        for key in self.supporting_languages:
+            if self.supporting_languages[key]['google_dest'] == dest:
+                return key
+        
+        raise RuntimeError(f"Detected langauge {dest} is not supported for TTS.")
+
+    def translate(self, query, lang):
+
+        dest = self._get_dest_from_lang(lang)
+
+        response = self.translator.translate_text(
+            request={
+                "parent": self.parent,
+                "contents": [query],
+                "mime_type": "text/plain",  # mime types: text/plain, text/html
+                "target_language_code": dest,
+            }
+        )
+
+        return " ".join([translation.translated_text for translation in response.translations])
+
+    def detect(self, query):
+        dest = self._detect(query)
+        return self._get_lang_from_dest(dest)