<commit_before>package org.mrgeo.mapalgebra

import java.awt.image.{DataBuffer, Raster}
import java.io.{Externalizable, IOException, ObjectInput, ObjectOutput}

import org.apache.spark.rdd.PairRDDFunctions
import org.apache.spark.{SparkConf, SparkContext}
import org.mrgeo.data.raster.RasterWritable
import org.mrgeo.data.rdd.RasterRDD
import org.mrgeo.image.MrsPyramidMetadata
import org.mrgeo.job.JobArguments
import org.mrgeo.mapalgebra.parser.{ParserException, ParserNode}
import org.mrgeo.mapalgebra.raster.RasterMapOp
import org.mrgeo.quantiles.Quantiles

import scala.language.existentials


object QuantilesMapOp extends MapOpRegistrar {
  override def register: Array[String] = {
    Array[String]("quantiles")
  }

  def create(raster: RasterMapOp, numQuantiles: Int, fraction:Float = 1.0f) =
    new QuantilesMapOp(Some(raster), Some(numQuantiles), Some(fraction))

  override def apply(node:ParserNode, variables: String => Option[ParserNode]): MapOp =
    new QuantilesMapOp(node, variables)
}

class QuantilesMapOp extends RasterMapOp with Externalizable {
  private var rasterRDD: Option[RasterRDD] = None

  private var inputMapOp: Option[RasterMapOp] = None
  private var numQuantiles: Option[Int] = None
  private var fraction: Option[Float] = None

  private[mapalgebra] def this(raster:Option[RasterMapOp], numQuantiles:Option[Int],
                               fraction: Option[Float]) = {
    this()

    this.inputMapOp = raster
    this.numQuantiles = numQuantiles
    this.fraction = fraction
  }

  def this(node: ParserNode, variables: String => Option[ParserNode]) {
    this()

    if ((node.getNumChildren < 2) || (node.getNumChildren > 3)) {
      throw new ParserException(
        "quantiles usage: quantiles(source raster, num quantiles, [percent of pixels to use])")
    }

    inputMapOp = RasterMapOp.decodeToRaster(node.getChild(0), variables)
    numQuantiles = MapOp.decodeInt(node.getChild(1), variables)
    if (numQuantiles.isEmpty) {
      throw new ParserException("The value for the numQuantiles parameter must be an integer")
    }
    if (node.getNumChildren > 2) {
      fraction = MapOp.decodeFloat(node.getChild(2), variables)
      fraction match {
        case Some(f) => {
          if ((f <= 0.0 || f > 1.0)) {
            throw new ParserException(
              "The fraction passed to quantiles " + f + " must be between 0.0 and 1.0")
          }
        }
        case None => throw new ParserException(
          "The value for the fraction parameter must be a number between 0.0 and 1.0");
      }
    }
  }

  override def rdd(): Option[RasterRDD] = {
    rasterRDD
  }

  override def registerClasses(): Array[Class[_]] = {
    Array[Class[_]](classOf[Array[Double]],
      classOf[Array[Float]],
      classOf[Array[Int]],
      classOf[Array[Short]],
      classOf[Array[Byte]],
      classOf[Array[Object]]
    )
  }

  override def execute(context: SparkContext): Boolean = {

    implicit val doubleOrdering = new Ordering[Double] {
      override def compare(x: Double, y: Double): Int = x.compareTo(y)
    }

    implicit val floatOrdering = new Ordering[Float] {
      override def compare(x: Float, y: Float): Int = x.compareTo(y)
    }

    implicit val intOrdering = new Ordering[Int] {
      override def compare(x: Int, y: Int): Int = x.compareTo(y)
    }

    implicit val shortOrdering = new Ordering[Short] {
      override def compare(x: Short, y: Short): Int = x.compareTo(y)
    }

    implicit val byteOrdering = new Ordering[Byte] {
      override def compare(x: Byte, y: Byte): Int = x.compareTo(y)
    }

    val input:RasterMapOp = inputMapOp getOrElse(throw new IOException("Input MapOp not valid!"))
    val numberOfQuantiles = numQuantiles getOrElse(throw new IOException("numQuantiles not valid!"))

    val meta = input.metadata() getOrElse(throw new IOException("Can't load metadata! Ouch! " + input.getClass.getName))
    rasterRDD = input.rdd()
    val rdd = rasterRDD getOrElse(throw new IOException("Can't load RDD! Ouch! " + inputMapOp.getClass.getName))
    // No reason to calculate metadata like raster map ops that actually compute a raster. This
    // map op does not compute the raster output, it just uses the input raster. All we need to
    // do is update the metadata already computed for the input raster map op.
//    metadata(SparkUtils.calculateMetadata(rasterRDD.get, meta.getMaxZoomLevel, meta.getDefaultValues,
//      bounds = meta.getBounds, calcStats = false))

    // Compute the quantile values and save them to metadata
    val quantiles = Quantiles.compute(rdd, numberOfQuantiles, fraction, meta)
    var b: Int = 0
    while (b < meta.getBands) {
      if (quantiles.contains(b)) {
        val quantileValues = quantiles(b)
        meta.setQuantiles(b, quantileValues)
      }
      b += 1
    }
    metadata(meta)
    true
  }

  override def setup(job: JobArguments, conf: SparkConf): Boolean = {
    true
  }

  override def teardown(job: JobArguments, conf: SparkConf): Boolean = {
    true
  }

  override def readExternal(in: ObjectInput): Unit = {}

  override def writeExternal(out: ObjectOutput): Unit = {}
}<commit_msg>Fix setting metadata with quantiles<commit_after>package org.mrgeo.mapalgebra

import java.awt.image.{DataBuffer, Raster}
import java.io.{Externalizable, IOException, ObjectInput, ObjectOutput}

import org.apache.spark.rdd.PairRDDFunctions
import org.apache.spark.{SparkConf, SparkContext}
import org.mrgeo.data.raster.RasterWritable
import org.mrgeo.data.rdd.RasterRDD
import org.mrgeo.image.MrsPyramidMetadata
import org.mrgeo.job.JobArguments
import org.mrgeo.mapalgebra.parser.{ParserException, ParserNode}
import org.mrgeo.mapalgebra.raster.RasterMapOp
import org.mrgeo.quantiles.Quantiles

import scala.language.existentials


object QuantilesMapOp extends MapOpRegistrar {
  override def register: Array[String] = {
    Array[String]("quantiles")
  }

  def create(raster: RasterMapOp, numQuantiles: Int, fraction:Float = 1.0f) =
    new QuantilesMapOp(Some(raster), Some(numQuantiles), Some(fraction))

  override def apply(node:ParserNode, variables: String => Option[ParserNode]): MapOp =
    new QuantilesMapOp(node, variables)
}

class QuantilesMapOp extends RasterMapOp with Externalizable {
  private var rasterRDD: Option[RasterRDD] = None

  private var inputMapOp: Option[RasterMapOp] = None
  private var numQuantiles: Option[Int] = None
  private var fraction: Option[Float] = None

  private[mapalgebra] def this(raster:Option[RasterMapOp], numQuantiles:Option[Int],
                               fraction: Option[Float]) = {
    this()

    this.inputMapOp = raster
    this.numQuantiles = numQuantiles
    this.fraction = fraction
  }

  def this(node: ParserNode, variables: String => Option[ParserNode]) {
    this()

    if ((node.getNumChildren < 2) || (node.getNumChildren > 3)) {
      throw new ParserException(
        "quantiles usage: quantiles(source raster, num quantiles, [percent of pixels to use])")
    }

    inputMapOp = RasterMapOp.decodeToRaster(node.getChild(0), variables)
    numQuantiles = MapOp.decodeInt(node.getChild(1), variables)
    if (numQuantiles.isEmpty) {
      throw new ParserException("The value for the numQuantiles parameter must be an integer")
    }
    if (node.getNumChildren > 2) {
      fraction = MapOp.decodeFloat(node.getChild(2), variables)
      fraction match {
        case Some(f) => {
          if ((f <= 0.0 || f > 1.0)) {
            throw new ParserException(
              "The fraction passed to quantiles " + f + " must be between 0.0 and 1.0")
          }
        }
        case None => throw new ParserException(
          "The value for the fraction parameter must be a number between 0.0 and 1.0");
      }
    }
  }

  override def rdd(): Option[RasterRDD] = {
    rasterRDD
  }

  override def registerClasses(): Array[Class[_]] = {
    Array[Class[_]](classOf[Array[Double]],
      classOf[Array[Float]],
      classOf[Array[Int]],
      classOf[Array[Short]],
      classOf[Array[Byte]],
      classOf[Array[Object]]
    )
  }

  override def execute(context: SparkContext): Boolean = {

    implicit val doubleOrdering = new Ordering[Double] {
      override def compare(x: Double, y: Double): Int = x.compareTo(y)
    }

    implicit val floatOrdering = new Ordering[Float] {
      override def compare(x: Float, y: Float): Int = x.compareTo(y)
    }

    implicit val intOrdering = new Ordering[Int] {
      override def compare(x: Int, y: Int): Int = x.compareTo(y)
    }

    implicit val shortOrdering = new Ordering[Short] {
      override def compare(x: Short, y: Short): Int = x.compareTo(y)
    }

    implicit val byteOrdering = new Ordering[Byte] {
      override def compare(x: Byte, y: Byte): Int = x.compareTo(y)
    }

    val input:RasterMapOp = inputMapOp getOrElse(throw new IOException("Input MapOp not valid!"))
    val numberOfQuantiles = numQuantiles getOrElse(throw new IOException("numQuantiles not valid!"))

    val meta = input.metadata() getOrElse(throw new IOException("Can't load metadata! Ouch! " + input.getClass.getName))
    rasterRDD = input.rdd()
    val rdd = rasterRDD getOrElse(throw new IOException("Can't load RDD! Ouch! " + inputMapOp.getClass.getName))
    // No reason to calculate metadata like raster map ops that actually compute a raster. This
    // map op does not compute the raster output, it just uses the input raster. All we need to
    // do is update the metadata already computed for the input raster map op.
//    metadata(SparkUtils.calculateMetadata(rasterRDD.get, meta.getMaxZoomLevel, meta.getDefaultValues,
//      bounds = meta.getBounds, calcStats = false))

    // Compute the quantile values and save them to metadata
    val quantiles = Quantiles.compute(rdd, numberOfQuantiles, fraction, meta)
    var b: Int = 0
    while (b < quantiles.length) {
      meta.setQuantiles(b, quantiles(b))
      b += 1
    }
    metadata(meta)
    true
  }

  override def setup(job: JobArguments, conf: SparkConf): Boolean = {
    true
  }

  override def teardown(job: JobArguments, conf: SparkConf): Boolean = {
    true
  }

  override def readExternal(in: ObjectInput): Unit = {}

  override def writeExternal(out: ObjectOutput): Unit = {}
}
{'fraction_non_alphanumeric': 0.10616705698672912, 'fraction_numerical': 0.00429352068696331, 'mean_word_length': 3.479458041958042, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 17, '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': '3961981', 'n_tokens_mistral': 3215, 'n_tokens_neox': 3113, 'n_words': 863}