<commit_before>package brainflight.binary

import akka.actor.Actor
import brainflight.tools.geometry.Point3D
import models.binary._
import scala.collection.mutable.ArrayBuffer
import akka.agent.Agent
import play.api.Logger
import play.api.libs.concurrent.Akka
import play.api.Play.current
import akka.actor._
import play.api.libs.concurrent.Promise
import play.api.libs.concurrent.Execution.Implicits._
import akka.actor.ActorSystem
import brainflight.tools.geometry.Vector3D
import brainflight.tools.Math._
import akka.pattern.ask
import akka.pattern.AskTimeoutException
import akka.util.Timeout
import scala.util._
import scala.concurrent.duration._
import scala.concurrent.Future
import akka.pattern.pipe
import com.typesafe.config.ConfigFactory
import play.api.Play
import brainflight.ActorSystems
import scala.collection.immutable.HashMap
import akka.routing.RoundRobinRouter

case class SingleRequest(dataRequest: DataRequest)
case class MultiCubeRequest(requests: Seq[SingleRequest])

class DataRequestActor extends Actor with DataCache {
  import DataStore._

  implicit val dataBlockLoadTimeout = Timeout(10 seconds)
  val conf = Play.current.configuration
  val remotePath = conf.getString("datarequest.remotepath").getOrElse("")
  val useRemote = conf.getBoolean("bindata.useRemote").getOrElse(false)
  implicit val system =
    if (useRemote)
      ActorSystems.dataRequestSystem
    else
      context.system

  lazy val dataStores = List[ActorRef](
    actorForWithLocalFallback[FileDataStore]("fileDataStore"),
    actorForWithLocalFallback[GridDataStore]("gridDataStore"),
    system.actorOf(Props(new EmptyDataStore()), name = "emptyDataStore"))

  def actorForWithLocalFallback[T <: Actor](name: String)(implicit evidence: scala.reflect.ClassTag[T]) = {
    if (useRemote)
      system.actorFor(s"$remotePath/user/$name")
    else
      system.actorOf(Props[T], name = name)
  }

  def receive = {
    case SingleRequest(dataRequest) =>
      load(dataRequest) pipeTo sender
    case MultiCubeRequest(requests) =>
      val resultsPromise = Future.traverse(requests)(r =>
        load(r.dataRequest))
      val s = sender
      resultsPromise.onComplete {
        case Success(results) =>
          val size = results.map(_.size).sum
          s ! Some(results.foldLeft(new ArrayBuffer[Byte](size))(_ ++= _))
        case Failure(e) =>
          Logger.error(s"DataRequestActor Error for Request. Error: $e")
          s ! None
      }
  }

  def loadFromSomewhere(dataSet: DataSet, dataLayer: DataLayer, resolution: Int, block: Point3D) = {
    val block2Load = LoadBlock(dataSet.baseDir, dataSet.name, dataLayer.folder, dataLayer.bytesPerElement, resolution, block.x, block.y, block.z)

    def loadFromStore(dataStores: List[ActorRef]): Future[Array[Byte]] = dataStores match {
      case a :: tail =>
        Logger.trace(s"Sending request: $block to ${a.path}")
        (a ? block2Load).mapTo[Array[Byte]].recoverWith {
          case e: AskTimeoutException =>
            Logger.warn(s"(${dataSet.name}/${dataLayer.folder} $block) ${a.path}: Not response in time.")
            loadFromStore(tail)
          case e: ClassCastException =>
            // TODO: find a better way to catch the DataNotFoundException
            Logger.warn(s"(${dataSet.name}/${dataLayer.folder} $block) ${a.path}: Not found.")
            loadFromStore(tail)
        }
      case _ =>
        throw new DataNotFoundException("DataSetActor")
    }

    withCache(block2Load) {
      loadFromStore(dataStores)
    }
  }

  def pointToBlock(point: Point3D, resolution: Int) =
    Point3D(
      point.x / blockLength / resolution,<commit_msg>Added exception to error message<commit_after>package brainflight.binary

import akka.actor.Actor
import brainflight.tools.geometry.Point3D
import models.binary._
import scala.collection.mutable.ArrayBuffer
import akka.agent.Agent
import play.api.Logger
import play.api.libs.concurrent.Akka
import play.api.Play.current
import akka.actor._
import play.api.libs.concurrent.Promise
import play.api.libs.concurrent.Execution.Implicits._
import akka.actor.ActorSystem
import brainflight.tools.geometry.Vector3D
import brainflight.tools.Math._
import akka.pattern.ask
import akka.pattern.AskTimeoutException
import akka.util.Timeout
import scala.util._
import scala.concurrent.duration._
import scala.concurrent.Future
import akka.pattern.pipe
import com.typesafe.config.ConfigFactory
import play.api.Play
import brainflight.ActorSystems
import scala.collection.immutable.HashMap
import akka.routing.RoundRobinRouter

case class SingleRequest(dataRequest: DataRequest)
case class MultiCubeRequest(requests: Seq[SingleRequest])

class DataRequestActor extends Actor with DataCache {
  import DataStore._

  implicit val dataBlockLoadTimeout = Timeout(10 seconds)
  val conf = Play.current.configuration
  val remotePath = conf.getString("datarequest.remotepath").getOrElse("")
  val useRemote = conf.getBoolean("bindata.useRemote").getOrElse(false)
  implicit val system =
    if (useRemote)
      ActorSystems.dataRequestSystem
    else
      context.system

  lazy val dataStores = List[ActorRef](
    actorForWithLocalFallback[FileDataStore]("fileDataStore"),
    actorForWithLocalFallback[GridDataStore]("gridDataStore"),
    system.actorOf(Props(new EmptyDataStore()), name = "emptyDataStore"))

  def actorForWithLocalFallback[T <: Actor](name: String)(implicit evidence: scala.reflect.ClassTag[T]) = {
    if (useRemote)
      system.actorFor(s"$remotePath/user/$name")
    else
      system.actorOf(Props[T], name = name)
  }

  def receive = {
    case SingleRequest(dataRequest) =>
      load(dataRequest) pipeTo sender
    case MultiCubeRequest(requests) =>
      val resultsPromise = Future.traverse(requests)(r =>
        load(r.dataRequest))
      val s = sender
      resultsPromise.onComplete {
        case Success(results) =>
          val size = results.map(_.size).sum
          s ! Some(results.foldLeft(new ArrayBuffer[Byte](size))(_ ++= _))
        case Failure(e) =>
          Logger.error(s"DataRequestActor Error for Request. Error: $e")
          s ! None
      }
  }

  def loadFromSomewhere(dataSet: DataSet, dataLayer: DataLayer, resolution: Int, block: Point3D) = {
    val block2Load = LoadBlock(dataSet.baseDir, dataSet.name, dataLayer.folder, dataLayer.bytesPerElement, resolution, block.x, block.y, block.z)

    def loadFromStore(dataStores: List[ActorRef]): Future[Array[Byte]] = dataStores match {
      case a :: tail =>
        Logger.trace(s"Sending request: $block to ${a.path}")
        (a ? block2Load).mapTo[Array[Byte]].recoverWith {
          case e: AskTimeoutException =>
            Logger.warn(s"(${dataSet.name}/${dataLayer.folder} $block) ${a.path}: Not response in time.")
            loadFromStore(tail)
          case e: ClassCastException =>
            // TODO: find a better way to catch the DataNotFoundException
            Logger.warn(s"(${dataSet.name}/${dataLayer.folder} $block) ${a.path}: Not found. E: $e")
            loadFromStore(tail)
        }
      case _ =>
        throw new DataNotFoundException("DataSetActor")
    }

    withCache(block2Load) {
      loadFromStore(dataStores)
    }
  }

  def pointToBlock(point: Point3D, resolution: Int) =
    Point3D(
      point.x / blockLength / resolution,
{'fraction_non_alphanumeric': 0.10832991382847763, 'fraction_numerical': 0.002735603884557516, 'mean_word_length': 3.813693219223173, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, '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': '4545786', 'n_tokens_mistral': 2210, 'n_tokens_neox': 2163, 'n_words': 463}