annotator.py (7972B)
1 from typing import Any, Dict, List, Set, Tuple, Union 2 import itertools 3 import os 4 import pathlib 5 import random 6 7 import IPython.display 8 import ipywidgets 9 10 import mmsrl.dataset 11 import mmsrl.utils 12 13 14 SAMPLE_PER_LABEL = 25 # Total of 100 annotations 15 SEED = 0 # We don't want to cherry pick what we annotate… 16 #PREFIX = pathlib.Path(os.environ["DATA_PATH"]) / "mmsrl" # we can put annotations in data 17 PREFIX = pathlib.Path(".") # The current directory (annotations in the git repository) seems better (we can commit the annotations) 18 SELECTED_PATH = PREFIX / "selected for annotation" 19 20 21 def get_data() -> Tuple[Dict[str, Any], List[Set[Tuple[str, str]]]]: 22 """ 23 Get the sample needing annotations split into the 4 labels. 24 25 Returns: 26 - the dataset, a image→sample dictionary 27 - sets of (image, entity) pairs grouped by label 28 """ 29 datasets = {} 30 for split in ["train", "val", "test"]: 31 config = mmsrl.utils.dotdict(dataname="all", model="none") 32 datasets[split] = mmsrl.dataset.MemeDataset(config, split, is_eval=True) 33 data = {sample["image_name"]: sample for sample in itertools.chain(*datasets.values())} 34 annotable = [set() for _ in range(4)] 35 for image, sample in data.items(): 36 for entity, label in zip(sample["entities_name"], sample["labels"]): 37 annotable[label].add((image, entity)) 38 return data, annotable 39 40 41 def select_annotations(annotable: List[Set[Tuple[str, str]]]): 42 """ Write the keys to the selected file if it does not exist. """ 43 if SELECTED_PATH.exists(): 44 raise RuntimeError("The file containing selected samples already exists, some people might have started annotation, don't change it now!") 45 rng = random.Random(SEED) 46 selected: List[Tuple[str, str]] = [] 47 for label in annotable: 48 domain: List[Tuple[str, str]] = sorted(label) # to ensure determinism 49 selected.extend(rng.sample(domain, SAMPLE_PER_LABEL)) 50 rng.shuffle(selected) # we don't want to present all victim in a row, etc. 51 with SELECTED_PATH.open("w") as selected_file: 52 for image, entity in selected: 53 print(f"{image}\t{entity}", file=selected_file) 54 55 56 class Annotator: 57 def __init__(self, identity: str, image_prefix: Union[pathlib.Path, str]): 58 self.identity = identity 59 self.image_prefix = pathlib.Path(image_prefix) 60 self.annotated_path = PREFIX / f"annotations of {self.identity}" 61 self.data, _ = get_data() 62 63 def instructions(self): 64 IPython.display.display(ipywidgets.HTML(f""" 65 <p><strong>This notebook is for annotator {self.identity}!</strong></p> 66 <p><strong>Instructions:</strong></p> 67 <ul> 68 <li>Given a meme and an entity, determine the role of the entity in the meme: hero vs. villain vs. victim vs. other.</li> 69 <li>The meme is to be analyzed from the perspective of the author of the meme.</li> 70 <li>Class definitions:<dl> 71 <dt>Hero</dt> 72 <dd>The entity is presented in a positive light. Glorified for their actions conveyed via the meme or gathered from background context.</dd> 73 <dt>Villain</td> 74 <dd>The entity is portrayed negatively, e.g., in an association with adverse traits like wickedness, cruelty, hypocrisy, etc.</dd> 75 <dt>Victim</dt> 76 <dd>The entity is portrayed as suffering the negative impact of someone else’s actions or conveyed implicitly within the meme.</dd> 77 <dt>Other</dt> 78 <dd>The entity is not a hero, a villain, or a victim.</dd> 79 </dl></li> 80 <li>The four classes are balanced, you should give each answer roughly 25% of the time.</li> 81 <li>Your answers are saved at each annotation, you can leave and come back at any time.</li> 82 </ul> 83 """)) 84 85 def get_already_annotated(self) -> Set[Tuple[str, str]]: 86 """ Get list of samples already annotated. """ 87 annotated = set() 88 try: 89 with self.annotated_path.open("r") as annotated_file: 90 for line in annotated_file: 91 image, entity, _ = line.rstrip('\n').split('\t') 92 annotated.add((image, entity)) 93 except FileNotFoundError: 94 pass 95 return annotated 96 97 def get_next_annotation(self) -> Tuple[str, str]: 98 """ Get the next (image, entity) pair to annotate. """ 99 # We re-read the file for each annotation to avoid an error-prone in-memory updating of the list 100 annotated = self.get_already_annotated() 101 # It would be hard to be less algorithmicaly efficient, but better be safe than sorry 102 with SELECTED_PATH.open("r") as selected_file: 103 for i, line in enumerate(selected_file): 104 image, entity = line.rstrip('\n').split('\t') 105 if (image, entity) not in annotated: 106 return i, image, entity 107 return -1, None, None 108 109 def write_answer(self, image: str, entity: str, answer: int): 110 with self.annotated_path.open("a") as annotated_file: 111 print(f"{image}\t{entity}\t{answer}", file=annotated_file) 112 113 def get_sample_html(self, image: str, entity: str): 114 sample = self.data[image] 115 ocr = sample["raw_text"] 116 html = f""" 117 <img src="{self.image_prefix / image}"/> 118 <pre>{ocr}</pre> 119 """ 120 return ipywidgets.HTML(html) 121 122 def delete_last_annotation(self): 123 annotations = [] 124 try: 125 with self.annotated_path.open("r") as annotated_file: 126 annotations = annotated_file.readlines() 127 except FileNotFoundError: 128 pass 129 if annotations: 130 annotations.pop() 131 with self.annotated_path.open("w") as annotated_file: 132 annotated_file.writelines(annotations) 133 134 def answer_clicked(self, image: str, entity: str, label: int): 135 def callback(button): 136 IPython.display.clear_output() 137 if label == -1: 138 # Undo last answer 139 self.delete_last_annotation() 140 else: 141 assert(button.description == mmsrl.utils.LABELS[label]) 142 self.write_answer(image, entity, label) 143 self.ask_next() 144 return callback 145 146 def ask_annotation(self, current_id: int, image: str, entity: str): 147 sample = self.get_sample_html(image, entity) 148 buttons = [] 149 for ilabel, label in enumerate(mmsrl.utils.LABELS): 150 button = ipywidgets.Button(description=label) 151 button.on_click(self.answer_clicked(image, entity, ilabel)) 152 buttons.append(button) 153 button = ipywidgets.Button(description="Undo last answer") 154 button.on_click(self.answer_clicked(None, None, -1)) 155 buttons.append(button) 156 hbox = ipywidgets.HBox(buttons) 157 instructions = ipywidgets.HTML(f'({current_id+1}/{4*SAMPLE_PER_LABEL}) <strong>What is the role of <span style="font-size:150%;">{entity}</span> in the above meme?</strong>') 158 vbox = ipywidgets.VBox([sample, instructions, hbox]) 159 IPython.display.display(vbox) 160 161 def completed(self): 162 message = ipywidgets.HTML(f'You completed all annotations! Thank you!') 163 button = ipywidgets.Button(description="Undo last answer") 164 button.on_click(self.answer_clicked(None, None, -1)) 165 vbox = ipywidgets.VBox([message, button]) 166 IPython.display.display(vbox) 167 168 def ask_next(self): 169 current_id, image, entity = self.get_next_annotation() 170 if current_id < 0 and image is None and entity is None: 171 self.completed() 172 else: 173 self.ask_annotation(current_id, image, entity) 174 175 176 # This only need to be done once for all annotators 177 # And it was already done, DO NOT UNCOMMENT 178 # select_annotations(get_data()[1])