<commit_before>package turkey
package tasks

import turkey.util._

import com.amazonaws.mturk.requester.AssignmentStatus

import scala.util.{Try, Success, Failure}
import scala.concurrent.duration._
import scala.language.postfixOps

import akka.actor.Actor
import akka.actor.ActorRef

import upickle.default._

import com.typesafe.scalalogging.StrictLogging

/**
  * Manages a particular kind of task; corresponds to a single TaskSpecification / HIT Type.
  * In here will be all of the logic related to how to review HITs, do quality control, keep track of auxiliary data,
  * schedule which HITs should be uploaded when, etc.
  */
abstract class HITManager[Prompt, Response](
  helper: HITManager.Helper[Prompt, Response]
) extends Actor {

  import helper.Message._

  final override def receive = receiveHelperMessage orElse receiveAux

  // delegates to helper when given a standard message defined in the helper
  private[this] final val receiveHelperMessage: PartialFunction[Any, Unit] = {
    case DisableAll => helper.disableAll
    case ReviewHITs => reviewHITs
    case AddPrompt(p) => addPrompt(p)
  }

  /** Override to add more incoming message types and message-processing logic */
  def receiveAux: PartialFunction[Any, Unit] =
    PartialFunction.empty[Any, Unit]

  /** Queries Turk and refreshes the task state, sending assignments for approval/validation,
    * approving/rejecting them, disposing HITs, etc. as necessary */
  def reviewHITs: Unit

  /** Adds a prompt to the set of prompts that this HITManager should be responsible for sourcing responses for. */
  def addPrompt(prompt: Prompt): Unit
}

object HITManager {
  /** Manages the ongoing state for a task with a particular HIT type;
    * keeps track of HITs and assignments that are active, saved, etc.
    * and gives convenience methods for interfacing with Turk. */
  class Helper[P, R](
    val taskSpec: TaskSpecification { type Prompt = P ; type Response = R })(
    implicit val promptReader: Reader[P],
    val responseReader: Reader[R],
    val responseWriter: Writer[R],
    val config: TaskConfig
  ) extends StrictLogging {
    private type Prompt = P
    private type Response = R

    import scala.collection.mutable

    object Message {
      sealed trait Message
      case object DisableAll extends Message
      case object ReviewHITs extends Message
      case class AddPrompt(prompt: Prompt) extends Message
    }
    import Message._
    import taskSpec.hitTypeId

    // disable method, not really complete yet

    def disableAll: Unit = {
      val currentlyActiveHITs = activeHITs.iterator.toList
      currentlyActiveHITs.foreach(disableHIT)
    }

    // HITs Active stuff

    // active HITs are currently up on Turk
    private[this] val activeHITs = {
      val active = mutable.Set.empty[HIT[Prompt]]
      for {
        mTurkHIT <- config.service.searchAllHITs
        if mTurkHIT.getHITTypeId.equals(hitTypeId)
        hit <- config.hitDataService.getHIT[Prompt](hitTypeId, mTurkHIT.getHITId).toOptionLogging(logger)
      } yield (active += hit)
      active
    }

    // finished means the HIT is not on turk (i.e., all assignments are done)
    // active includes HITs for which some assignments are done and some are not
    private[this] val (finishedHITInfosByPrompt, activeHITInfosByPrompt) = {
      val finishedRes = mutable.Map.empty[Prompt, List[HITInfo[Prompt, Response]]]
      val activeRes = mutable.Map.empty[Prompt, List[HITInfo[Prompt, Response]]]
      config.hitDataService.getAllHITInfo[Prompt, Response](hitTypeId).get
        .groupBy(_.hit.prompt)
        .foreach { case (prompt, infos) =>
          infos.foreach { hitInfo =>
            if(activeHITs.contains(hitInfo.hit)) {
              activeRes.put(prompt, hitInfo :: activeRes.get(prompt).getOrElse(Nil))
            } else {
              finishedRes.put(prompt, hitInfo :: activeRes.get(prompt).getOrElse(Nil))
            }
          }
      }
      (finishedRes, activeRes)
    }

    def finishedHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      finishedHITInfosByPrompt.iterator
    def finishedHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      finishedHITInfosByPrompt.get(p).getOrElse(Nil)
    def activeHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      activeHITInfosByPrompt.iterator
    def activeHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      activeHITInfosByPrompt.get(p).getOrElse(Nil)
    def allCurrentHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      activeHITInfosByPromptIterator ++ finishedHITInfosByPromptIterator
    def allCurrentHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      activeHITInfos(p) ++ finishedHITInfos(p)

    /** Create a HIT with the specific parameters.
      * This should be used in order to ensure the helper has a consistent state.
      */
    def createHIT(prompt: Prompt, numAssignments: Int): Try[HIT[Prompt]] = {
      val attempt = taskSpec.createHIT(prompt, numAssignments)
      attempt match {
        case Success(hit) =>
          activeHITs += hit
          val newHITInfo = HITInfo[Prompt, Response](hit, Nil)
          activeHITInfosByPrompt.put(prompt, newHITInfo :: activeHITInfos(prompt))
          logger.info(s"Created HIT: ${hit.hitId}\n${config.service.getWebsiteURL}/mturk/preview?groupId=${hit.hitTypeId}")
        case Failure(e) =>
          logger.error(e.getMessage)
          e.printStackTrace
      }
      attempt
    }

    def isActive(prompt: Prompt): Boolean = activeHITInfosByPrompt.contains(prompt)
    def isActive(hit: HIT[Prompt]): Boolean = activeHITs.contains(hit)
    def isActive(hitId: String): Boolean = activeHITs.exists(_.hitId == hitId)
    def numActiveHITs = activeHITs.size

    private[this] def processRemovedHIT(hit: HIT[Prompt]): Unit = {
      if(!isActive(hit)) {
        logger.error(s"Trying to finish HIT that isn't active? $hit")
      } else {
        activeHITs -= hit
        // add to other appropriate data structures
        val finishedData = finishedHITInfos(hit.prompt)
        val activeData = activeHITInfos(hit.prompt)
        val curInfo = activeData
          .find(_.hit.hitId == hit.hitId)
          .getOrElse {
          logger.error("Could not find active HIT to move to finished");
          HITInfo(
            hit,
            config.hitDataService.getAssignmentsForHIT[Response](hitTypeId, hit.hitId).get)
        }
        val newActiveData = activeData.filterNot(_.hit.hitId == hit.hitId)
        val newFinishedData = curInfo :: finishedData
        activeHITInfosByPrompt.put(hit.prompt, newActiveData)
        finishedHITInfosByPrompt.put(hit.prompt, newFinishedData)
      }
    }

    def disableHIT(hit: HIT[Prompt]): Unit = {
      Try(config.service.disableHIT(hit.hitId)) match {
        case Success(_) =>
          logger.info(s"Disabled HIT: ${hit.hitId}\nHIT type for disabled HIT: ${hitTypeId}")
          processRemovedHIT(hit)
        case Failure(e) =>
          logger.error(s"HIT disabling failed:\n$hit\n$e")
      }
    }

    /** Disposes of a disposable HIT and takes care of bookkeeping.
      * Assumes the HIT is disposable.
      */
    def finishHIT(hit: HIT[Prompt]): Unit = {
      Try(config.service.disposeHIT(hit.hitId)) match {
        case Success(_) =><commit_msg>Fix minor bug with tracking currently active prompts/HITs<commit_after>package turkey
package tasks

import turkey.util._

import com.amazonaws.mturk.requester.AssignmentStatus

import scala.util.{Try, Success, Failure}
import scala.concurrent.duration._
import scala.language.postfixOps

import akka.actor.Actor
import akka.actor.ActorRef

import upickle.default._

import com.typesafe.scalalogging.StrictLogging

/**
  * Manages a particular kind of task; corresponds to a single TaskSpecification / HIT Type.
  * In here will be all of the logic related to how to review HITs, do quality control, keep track of auxiliary data,
  * schedule which HITs should be uploaded when, etc.
  */
abstract class HITManager[Prompt, Response](
  helper: HITManager.Helper[Prompt, Response]
) extends Actor {

  import helper.Message._

  final override def receive = receiveHelperMessage orElse receiveAux

  // delegates to helper when given a standard message defined in the helper
  private[this] final val receiveHelperMessage: PartialFunction[Any, Unit] = {
    case DisableAll => helper.disableAll
    case ReviewHITs => reviewHITs
    case AddPrompt(p) => addPrompt(p)
  }

  /** Override to add more incoming message types and message-processing logic */
  def receiveAux: PartialFunction[Any, Unit] =
    PartialFunction.empty[Any, Unit]

  /** Queries Turk and refreshes the task state, sending assignments for approval/validation,
    * approving/rejecting them, disposing HITs, etc. as necessary */
  def reviewHITs: Unit

  /** Adds a prompt to the set of prompts that this HITManager should be responsible for sourcing responses for. */
  def addPrompt(prompt: Prompt): Unit
}

object HITManager {
  /** Manages the ongoing state for a task with a particular HIT type;
    * keeps track of HITs and assignments that are active, saved, etc.
    * and gives convenience methods for interfacing with Turk. */
  class Helper[P, R](
    val taskSpec: TaskSpecification { type Prompt = P ; type Response = R })(
    implicit val promptReader: Reader[P],
    val responseReader: Reader[R],
    val responseWriter: Writer[R],
    val config: TaskConfig
  ) extends StrictLogging {
    private type Prompt = P
    private type Response = R

    import scala.collection.mutable

    object Message {
      sealed trait Message
      case object DisableAll extends Message
      case object ReviewHITs extends Message
      case class AddPrompt(prompt: Prompt) extends Message
    }
    import Message._
    import taskSpec.hitTypeId

    // disable method, not really complete yet

    def disableAll: Unit = {
      val currentlyActiveHITs = activeHITs.iterator.toList
      currentlyActiveHITs.foreach(disableHIT)
    }

    // HITs Active stuff

    // active HITs are currently up on Turk
    private[this] val activeHITs = {
      val active = mutable.Set.empty[HIT[Prompt]]
      for {
        mTurkHIT <- config.service.searchAllHITs
        if mTurkHIT.getHITTypeId.equals(hitTypeId)
        hit <- config.hitDataService.getHIT[Prompt](hitTypeId, mTurkHIT.getHITId).toOptionLogging(logger)
      } yield (active += hit)
      active
    }

    // finished means the HIT is not on turk (i.e., all assignments are done)
    // active includes HITs for which some assignments are done and some are not
    private[this] val (finishedHITInfosByPrompt, activeHITInfosByPrompt) = {
      val finishedRes = mutable.Map.empty[Prompt, List[HITInfo[Prompt, Response]]]
      val activeRes = mutable.Map.empty[Prompt, List[HITInfo[Prompt, Response]]]
      config.hitDataService.getAllHITInfo[Prompt, Response](hitTypeId).get
        .groupBy(_.hit.prompt)
        .foreach { case (prompt, infos) =>
          infos.foreach { hitInfo =>
            if(activeHITs.contains(hitInfo.hit)) {
              activeRes.put(prompt, hitInfo :: activeRes.get(prompt).getOrElse(Nil))
            } else {
              finishedRes.put(prompt, hitInfo :: activeRes.get(prompt).getOrElse(Nil))
            }
          }
      }
      (finishedRes, activeRes)
    }

    def finishedHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      finishedHITInfosByPrompt.iterator
    def finishedHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      finishedHITInfosByPrompt.get(p).getOrElse(Nil)
    def activeHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      activeHITInfosByPrompt.iterator
    def activeHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      activeHITInfosByPrompt.get(p).getOrElse(Nil)
    def allCurrentHITInfosByPromptIterator: Iterator[(Prompt, List[HITInfo[Prompt, Response]])] =
      activeHITInfosByPromptIterator ++ finishedHITInfosByPromptIterator
    def allCurrentHITInfos(p: Prompt): List[HITInfo[Prompt, Response]] =
      activeHITInfos(p) ++ finishedHITInfos(p)

    /** Create a HIT with the specific parameters.
      * This should be used in order to ensure the helper has a consistent state.
      */
    def createHIT(prompt: Prompt, numAssignments: Int): Try[HIT[Prompt]] = {
      val attempt = taskSpec.createHIT(prompt, numAssignments)
      attempt match {
        case Success(hit) =>
          activeHITs += hit
          val newHITInfo = HITInfo[Prompt, Response](hit, Nil)
          activeHITInfosByPrompt.put(prompt, newHITInfo :: activeHITInfos(prompt))
          logger.info(s"Created HIT: ${hit.hitId}\n${config.service.getWebsiteURL}/mturk/preview?groupId=${hit.hitTypeId}")
        case Failure(e) =>
          logger.error(e.getMessage)
          e.printStackTrace
      }
      attempt
    }

    def isActive(prompt: Prompt): Boolean = activeHITInfosByPrompt.contains(prompt)
    def isActive(hit: HIT[Prompt]): Boolean = activeHITs.contains(hit)
    def isActive(hitId: String): Boolean = activeHITs.exists(_.hitId == hitId)
    def numActiveHITs = activeHITs.size

    private[this] def processRemovedHIT(hit: HIT[Prompt]): Unit = {
      if(!isActive(hit)) {
        logger.error(s"Trying to finish HIT that isn't active? $hit")
      } else {
        activeHITs -= hit
        // add to other appropriate data structures
        val finishedData = finishedHITInfos(hit.prompt)
        val activeData = activeHITInfos(hit.prompt)
        val curInfo = activeData
          .find(_.hit.hitId == hit.hitId)
          .getOrElse {
          logger.error("Could not find active HIT to move to finished");
          HITInfo(
            hit,
            config.hitDataService.getAssignmentsForHIT[Response](hitTypeId, hit.hitId).get)
        }
        val newActiveData = activeData.filterNot(_.hit.hitId == hit.hitId)
        val newFinishedData = curInfo :: finishedData
        if(newActiveData.isEmpty) {
          activeHITInfosByPrompt.remove(hit.prompt)
        } else {
          activeHITInfosByPrompt.put(hit.prompt, newActiveData)
        }
        finishedHITInfosByPrompt.put(hit.prompt, newFinishedData)
      }
    }

    def disableHIT(hit: HIT[Prompt]): Unit = {
      Try(config.service.disableHIT(hit.hitId)) match {
        case Success(_) =>
          logger.info(s"Disabled HIT: ${hit.hitId}\nHIT type for disabled HIT: ${hitTypeId}")
          processRemovedHIT(hit)
        case Failure(e) =>
          logger.error(s"HIT disabling failed:\n$hit\n$e")
      }
    }

    /** Disposes of a disposable HIT and takes care of bookkeeping.
      * Assumes the HIT is disposable.
      */
    def finishHIT(hit: HIT[Prompt]): Unit = {
      Try(config.service.disposeHIT(hit.hitId)) match {
        case Success(_) =>
{'fraction_non_alphanumeric': 0.09516788419006769, 'fraction_numerical': 0.0, 'mean_word_length': 3.171652222532849, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6885516', 'n_tokens_mistral': 4658, 'n_tokens_neox': 4529, 'n_words': 1172}