class_docstr: Class to compile results from static turn annotations.
  
  Change PROBLEM_BUCKETS in task_config/annotations_config.json to be the buckets that
  you are asking crowdsource workers to annotate with.
  
  NOTE this script directly accesses Mephisto files rather than using the DataBrowser.
  This makes it fragile and not a great example for extension. It is a good candidate
  for refactor.
class_docstr_tok: ["Class","to","compile","results","from","static","turn","annotations",".","Change","PROBLEM_BUCKETS","in","task_config\/annotations_config.json","to","be","the","buckets","that","you","are","asking","crowdsource","workers","to","annotate","with",".","NOTE","this","script","directly","accesses","Mephisto","files","rather","than","using","the","DataBrowser",".","This","makes","it","fragile","and","not","a","great","example","for","extension",".","It","is","a","good","candidate","for","refactor","."]
docstr: Do initial loading and processing of crowdsource data Loads data from all the
  worker ID files and gets rid of incomplete or malformed convos.
  
  Also adds fields such as worker_id, assignment_id, etc for convenience
  :return: list of JSON objects which represent a conversation with
  annotations d["data"] of each has an array of utterance level data
docstr_tok: ["Do","initial","loading","and","processing","of","crowdsource","data","Loads","data","from","all","the","worker","ID","files","and","gets","rid","of","incomplete","or","malformed","convos",".","Also","adds","fields","such","as","worker_id",",","assignment_id",",","etc","for","convenience",":","return",":","list","of","JSON","objects","which","represent","a","conversation","with","annotations","d","[","``","data","''","]","of","each","has","an","array","of","utterance","level","data"]
code: def compile_initial_results(self, results_folders) -> list:
      """
      Do initial loading and processing of crowdsource data Loads data from all the
      worker ID files and gets rid of incomplete or malformed convos.
  
      Also adds fields such as worker_id, assignment_id, etc for convenience
      :return: list of JSON objects which represent a conversation with
      annotations d["data"] of each has an array of utterance level data
      """
      print("Starting compile_initial_results...")
      all_data_paths = []
      for f in results_folders:
          # Each one is a HIT completed by a given worker (so if
          # units-per-assignment > 1), then will include the same conversations
          # multiple times annotated by different workers
          data_paths = self.get_data_paths_mephisto(f)
          all_data_paths.extend(data_paths)
      print(f"Got {len(all_data_paths)} folders to read.")
  
      conversations = []
      task_completion_times = []
      for dp in all_data_paths:
          # Read in file
          with open(os.path.join(dp), "rb") as f:
              data = json.load(f)
  
          worker_id = dp.split("/")[-2]
          hit_id = dp.split("/")[-3]
          _ = dp.split("/")[-4]  # Task run
  
          (is_valid_hit, reason) = self._validate_hit(data)
          if not is_valid_hit:
              print(
                  f"Skipping invalid HIT {hit_id}, worker_id: {worker_id} for reason: {reason}."
              )
              continue
  
          # HIT-level metric of HIT completion time has to be done here for now
          try:
              task_completion_time_seconds = (
                  data["times"]["task_end"]
                  - data["times"]["task_start"]
              )
          except KeyError:
              # We've been relying on non-mephisto API, and in 1.0.2 'times' was deprecated
              # so this uses the new location
              metadata_path = os.path.join(
                  os.path.dirname(dp), "agent_meta.json"
              )
              with open(metadata_path, "rb") as f:
                  metadata = json.load(f)
  
              # HIT-level metric of HIT completion time has to be done here for now
              task_completion_time_seconds = (
                  metadata["task_end"] - metadata["task_start"]
              )
          print(task_completion_time_seconds)
  
          subtasks = data["outputs"]["final_data"]
          if self.LIVE_ONBOARDING_IS_LAST_SUBTASK:
              qc_success_pct = (
                  self._get_inflight_onboarding_success_from_subtask(
                      subtasks[-1]
                  )
              )
          else:
              qc_success_pct = 0.0
          for subtask_idx, d in enumerate(subtasks):
              if (
                  subtask_idx == (len(subtasks) - 1)
                  and self.LIVE_ONBOARDING_IS_LAST_SUBTASK
              ):
                  # Last subtask is inflight onboarding; don't include it
                  continue
              subtask_data = copy.deepcopy(d)
              # Structure of each subtask is {'subtask_index': XX, 'data': [...]}
              (is_valid_subtask, reason) = self._validate_subtask(
                  d["data"]
              )
              if not is_valid_subtask:
                  print(
                      f"Skipping invalid subtask within HIT: {hit_id}, worker_id: {worker_id} for reason: {reason}."
                  )
                  continue
  
              subtask_data["worker_id"] = worker_id
              subtask_data["hit_id"] = hit_id
              subtask_data["folder"] = dp
              subtask_data["subtask_idx"] = subtask_idx
              experimental_design = "self_chat"
              subtask_data["model_nickname"] = (
                  experimental_design + "/" + "TODO"
              )
              subtask_data["qc_success_pct"] = qc_success_pct
              conversations.append(subtask_data)
  
          task_completion_times.append(task_completion_time_seconds)
  
      if len(task_completion_times) > 0:
          print(
              f"Average task completion time (seconds) was: "
              f"{np.average(task_completion_times):0.1f}"
          )
  
      if len(conversations) == 0:
          raise ValueError("No conversations found!")
  
      return conversations
  
imports: import copy
  import json
  import os
  from datetime import datetime
  from typing import Any, Dict, Optional, Tuple
  import numpy
  import pandas
  from parlai.crowdsourcing.utils.analysis import AbstractTurnAnnotationResultsCompiler
label_desc: Use image node_id 9 for calling the TurnAnnotationsStaticResultsCompiler obj's underlying member method code with example usage: obj.compile_initial_results(results_folders) and returns: conversations