Upload task.py with huggingface_hub

task.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from .operator import StreamInstanceOperator
 
@@ -11,22 +11,31 @@ class FormTask(Tasker, StreamInstanceOperator):
     inputs: List[str]
     outputs: List[str]
     metrics: List[str]
+    augmentable_inputs: List[str] = []
 
-    def process(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
+    def verify(self):
+        for augmentable_input in self.augmentable_inputs:
+            assert (
+                augmentable_input in self.inputs
+            ), f"augmentable_input f{augmentable_input} is not part of {self.inputs}"
+
+    def process(
+        self, instance: Dict[str, Any], stream_name: Optional[str] = None
+    ) -> Dict[str, Any]:
         try:
             inputs = {key: instance[key] for key in self.inputs}
         except KeyError as e:
             raise KeyError(
-                f"Unexpected input column names: {list(key for key in self.inputs if key not in instance)}"
-                f"\n available names:{list(instance.keys())}\n given input names:{self.inputs}"
-            )
+                f"Unexpected FormTask input column names ({[key for key in self.inputs if key not in instance]})."
+                f"The available input names: {list(instance.keys())}"
+            ) from e
         try:
             outputs = {key: instance[key] for key in self.outputs}
         except KeyError as e:
             raise KeyError(
-                f"Unexpected output column names: {list(key for key in self.outputs if key not in instance)}"
+                f"Unexpected FormTask output column names: {[key for key in self.outputs if key not in instance]}"
                 f" \n available names:{list(instance.keys())}\n given output names:{self.outputs}"
-            )
+            ) from e
 
         return {
             "inputs": inputs,
@@ -42,13 +51,15 @@ class MultipleChoiceTask(FormTask):
     use_text_in_target: bool = False
     alphabet: str = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
-    def process_single_choice(self, choice: str, index: int, use_text: bool = True) -> str:
+    def process_single_choice(
+        self, choice: str, index: int, use_text: bool = True
+    ) -> str:
         try:
             processed_choice = f"{self.alphabet[index]}"
-        except IndexError:
+        except IndexError as e:
             raise ValueError(
                 f"Too many choices, the length of alphabet '{self.alphabet}': {len(self.alphabet)} is the limit"
-            )
+            ) from e
         if use_text:
             processed_choice += f"{self.enumeration_suffix}{choice}"
         return processed_choice
@@ -60,9 +71,13 @@ class MultipleChoiceTask(FormTask):
         return self.choices_separator.join(processed_choices)
 
     def process_target(self, choices, target_index):
-        return self.process_single_choice(choices[target_index], target_index, use_text=self.use_text_in_target)
+        return self.process_single_choice(
+            choices[target_index], target_index, use_text=self.use_text_in_target
+        )
 
-    def process(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
+    def process(
+        self, instance: Dict[str, Any], stream_name: Optional[str] = None
+    ) -> Dict[str, Any]:
         result = super().process(instance, stream_name)
         target_key, target_value = next(iter(result["outputs"].items()))
         choices = result["inputs"][self.choices_field]