Add operation when updating

Author: jordimas

open_dubbing/utterance.py

@@ -185,48 +185,101 @@ def get_without_empty_blocks(self, utterance_metadata):
 
         return new_utterance
 
+    def _get_highest_id(self, utterance_metadata):
+        highest_id = 1
+        for utterance in utterance_metadata:
+            id = utterance["id"]
+            if id > highest_id:
+                highest_id = id
+        return highest_id
+
+    def _create_new_utterance(self, update, new_id):
+        mandatory_fields = [
+            "speaker_id",
+            "translated_text",
+            "assigned_voice",
+            "gender",
+            "start",
+            "end",
+        ]
+
+        new_utterance = {"id": new_id}
+
+        for field in mandatory_fields:
+            value = update.get(field, None)
+            if not value:
+                logger().warning(
+                    f"Missing field '{field}' when adding new utterance with id '{new_id}'"
+                )
+                return None
+
+            new_utterance[field] = value
+
+        return new_utterance
+
+    def _update_utterance(self, update, utterance):
+        updateable_fields = [
+            "speaker_id",
+            "translated_text",
+            "speed",
+            "assigned_voice",
+            "for_dubbing",
+            "gender",
+            "start",
+            "end",
+        ]
+        for field in updateable_fields:
+            value = update.get(field, None)
+            if not value:
+                continue
+
+            utterance[field] = value
+        return utterance
+
     def update_utterances(self, utterance_master, utterance_update):
-        id_to_update = {}
+        id_to_update_or_delete = {}
+        id_to_create = {}
         utterance_new = []
 
         for utterance in utterance_update:
             id = utterance["id"]
-            id_to_update[id] = utterance
+            operation = utterance["operation"]
+            if operation == "create":
+                if id == 0:
+                    new_id = self._get_highest_id(utterance_master) + 1
+                    new_utterance = self._create_new_utterance(utterance, new_id)
+                    if new_utterance:
+                        utterance_new.append(new_utterance)
+                else:
+                    id_to_create[id] = utterance
+            else:
+                id_to_update_or_delete[id] = utterance
 
         for utterance in utterance_master:
             id = utterance["id"]
-            update = id_to_update.get(id, None)
+
+            update = id_to_update_or_delete.get(id, None)
             if not update:
                 utterance_new.append(utterance)
-                continue
-
-            operation = update.get("operation", None)
-            if not operation:
-                raise ValueError("No operation field defined")
+            else:
+                operation = update.get("operation", None)
+                if not operation:
+                    raise ValueError("No operation field defined")
 
-            if operation == "delete":
-                continue
-
-            if operation != "update":
-                raise ValueError(f"Invalid operation {operation}")
-
-            updateable_fields = [
-                "speaker_id",
-                "translated_text",
-                "speed",
-                "assigned_voice",
-                "for_dubbing",
-                "gender",
-                "start",
-                "end",
-            ]
-            for field in updateable_fields:
-                value = update.get(field, None)
-                if not value:
+                if operation == "delete":
                     continue
 
-                utterance[field] = value
+                if operation != "update":
+                    raise ValueError(f"Invalid operation {operation}")
+
+                utterance = self._update_utterance(update, utterance)
+                utterance_new.append(utterance)
 
-            utterance_new.append(utterance)
+            create = id_to_create.get(id, None)
+            if create:
+                new_id = self._get_highest_id(utterance_master) + 1
+                new_utterance = self._create_new_utterance(create, new_id)
+                if new_utterance:
+                    utterance_new.append(new_utterance)
 
         return utterance_new

tests/utterance_test.py

@@ -23,7 +23,6 @@
 class TestUterrance:
 
     def testrun_save_utterance(self):
-
         with tempfile.TemporaryDirectory() as temp_dir:
             directory = temp_dir
 
@@ -65,7 +64,6 @@ def testrun_save_utterance(self):
                 }
 
     def test_hash_utterances(self):
-
         utterances = [
             {
                 "start": 1.26,
@@ -145,7 +143,6 @@ def test_get_without_empty_blocks(self):
         assert "Hola" == modified[0]["text"]
 
     def test_add_unique_ids(self):
-
         utterances = [
             {
                 "start": 1.26,
@@ -201,8 +198,73 @@ def _get_master_utterances(self):
             },
         ]
 
-    def test_update_utterances_operation_delete(self):
+    def test_update_utterances_operation_create(self):
+        master = self._get_master_utterances()
+        utterance = Utterance(
+            target_language="cat",
+            output_directory=None,
+        )
+        create_utterances = [
+            {
+                "id": 1,
+                "operation": "create",
+                "translated_text": "Bon dia",
+                "speaker_id": "SPEAKER_01",
+                "gender": "Male",
+                "assigned_voice": "ca-ES-EnricNeural",
+                "start": 4,
+                "end": 5,
+            }
+        ]
+        new_utterances = utterance.update_utterances(master, create_utterances)
+        assert len(new_utterances) == 3
+        assert [u["id"] for u in new_utterances] == [1, 3, 2]
+        assert new_utterances[1]["translated_text"] == "Bon dia"
 
+    def test_update_utterances_operation_create_first(self):
+        master = self._get_master_utterances()
+        utterance = Utterance(
+            target_language="cat",
+            output_directory=None,
+        )
+        create_utterances = [
+            {
+                "id": 0,
+                "operation": "create",
+                "translated_text": "Bon dia",
+                "speaker_id": "SPEAKER_01",
+                "gender": "Male",
+                "assigned_voice": "ca-ES-EnricNeural",
+                "start": 4,
+                "end": 5,
+            }
+        ]
+        new_utterances = utterance.update_utterances(master, create_utterances)
+        assert len(new_utterances) == 3
+        assert [u["id"] for u in new_utterances] == [3, 1, 2]
+        assert new_utterances[0]["translated_text"] == "Bon dia"
+
+    def test_update_utterances_operation_create_ignore_missing_fields(self):
+        master = self._get_master_utterances()
+        utterance = Utterance(
+            target_language="cat",
+            output_directory=None,
+        )
+        create_utterances = [
+            {
+                "id": 0,
+                "operation": "create",
+                "gender": "Male",
+                "assigned_voice": "ca-ES-EnricNeural",
+                "start": 4,
+                "end": 5,
+            }
+        ]
+        new_utterances = utterance.update_utterances(master, create_utterances)
+        assert len(new_utterances) == 2
+        assert [u["id"] for u in new_utterances] == [1, 2]
+
+    def test_update_utterances_operation_delete(self):
         master = self._get_master_utterances()
         utterance = Utterance(
             target_language="cat",