<commit_before>package io.github.mandar2812.PlasmaML.helios

import ammonite.ops.{Path, pwd, root}
import breeze.linalg.DenseVector
import com.sksamuel.scrimage.Image
import com.sksamuel.scrimage.filter.GrayscaleFilter
import io.github.mandar2812.PlasmaML.helios
import io.github.mandar2812.PlasmaML.helios.data.SDOData.Instruments._
import io.github.mandar2812.PlasmaML.omni.{OMNIData, OMNILoader}
import io.github.mandar2812.dynaml.pipes._
import io.github.mandar2812.dynaml.analysis.implicits._
import io.github.mandar2812.dynaml.probability.{DiscreteDistrRV, MultinomialRV}
import io.github.mandar2812.dynaml.tensorflow.{dtf, dtfdata, dtfpipe}
import io.github.mandar2812.dynaml.tensorflow.data.{AbstractDataSet, DataSet, TFDataSet, ZipDataSet}
import io.github.mandar2812.dynaml.tensorflow.utils.{GaussianScalerTF, MinMaxScalerTF}
import io.github.mandar2812.dynaml.DynaMLPipe._
import io.github.mandar2812.dynaml.utils
import org.joda.time._
import spire.math.UByte
import org.platanios.tensorflow.api._


/**
  * <h3>Helios Data Facility</h3>
  *
  * Package enabling downloading and storage and processing
  * of heliospheric image data.
  *
  * @author mandar2812
  * */
package object data {


  private implicit val dateOrdering = new Ordering[DateTime] {
    override def compare(x: DateTime, y: DateTime): Int = if(x.isBefore(y)) -1 else 1
  }

  type IMAGE_PATTERN           = (DateTime, Path)

  /**
    * A simple data pattern, consisting of
    * a time stamp, path to an image, and a sequence of numbers
    * */
  type PATTERN                 = (DateTime, (Path, Seq[Double]))

  /**
    * A pattern, consisting of
    * a time stamp, path to an image, a tuple of numeric sequences
    * */
  type PATTERN_EXT             = (DateTime, (Path, (Seq[Double], Seq[Double])))

  /**
    * A pattern, consisting of
    * a time stamp, a collection of images from multiple sources,
    * and a sequence of numbers
    * */
  type MC_PATTERN              = (DateTime, (Map[SolarImagesSource, Seq[Path]], Seq[Double]))

  /**
    * A pattern, consisting of
    * a time stamp, a collection of images from multiple sources,
    * and a tuple of sequence of numbers
    * */
  type MC_PATTERN_EXT          = (DateTime, (Map[SOHO, Seq[Path]], (Seq[Double], Seq[Double])))

  type HELIOS_IMAGE_DATA       = DataSet[IMAGE_PATTERN]
  type HELIOS_OMNI_DATA        = DataSet[PATTERN]
  type HELIOS_MC_OMNI_DATA     = DataSet[MC_PATTERN]
  type HELIOS_OMNI_DATA_EXT    = DataSet[PATTERN_EXT]
  type HELIOS_MC_OMNI_DATA_EXT = Iterable[MC_PATTERN_EXT]

  type IMAGE_TS                = (Tensor, Tensor)

  type TF_IMAGE_DATA           = TFDataSet[Tensor]

  type TF_DATA                 = TFDataSet[(Tensor, Tensor)]

  type TF_DATA_EXT             = TFDataSet[((Tensor, Tensor), Tensor)]

  type TF_MC_DATA_EXT          = AbstractDataSet[(Tensor, Tensor), Tensor]

  type SC_TF_MC_DATA_EXT       = (TF_MC_DATA_EXT, (ReversibleScaler[(Tensor, Tensor)], MinMaxScalerTF))

  type SC_TF_DATA_EXT          = (TF_DATA_EXT, (ReversibleScaler[(Tensor, Tensor)], MinMaxScalerTF))

  type SC_TF_DATA              = (TF_DATA, (ReversibleScaler[Tensor], MinMaxScalerTF))

  private def TF_DATA_EXT(
    trData: IMAGE_TS,
    trLabels: Tensor,
    sizeTr: Int,
    tData: IMAGE_TS,
    tLabels: Tensor,
    sizeT: Int): AbstractDataSet[IMAGE_TS, Tensor] =
    AbstractDataSet(trData, trLabels, sizeTr, tData, tLabels, sizeT)



  /**
    * Download a resource (image, file) from a sequence of urls to a specified
    * disk location.
    * */
  def download_batch(path: Path)(urls: List[String]): Unit = {
    urls.par.foreach(s => utils.downloadURL(s, (path/s.split('/').last).toString()))
  }

  /**
    * Perform a bulk download of images within some date range
    * */
  def download_day_range(download: LocalDate => Unit)(start: LocalDate, end: LocalDate): Unit = {

    val num_days = new Duration(start.toDateTimeAtStartOfDay, end.toDateTimeAtStartOfDay).getStandardDays.toInt

    (0 to num_days).map(start.plusDays).par.foreach(download)
  }

  def download_month_range(download: YearMonth => Unit)(start: YearMonth, end: YearMonth): Unit = {

    val period = new Period(
      start.toLocalDate(1).toDateTimeAtStartOfDay,
      end.toLocalDate(31).toDateTimeAtStartOfDay)

    val num_months = (12*period.getYears) + period.getMonths

    (0 to num_months).map(start.plusMonths).par.foreach(download)
  }

  sealed trait Source
  sealed trait SolarImagesSource extends Source {
    val instrument: String
    val size: Int
  }

  case class SOHO(
    override val instrument: String,
    override val size: Int = SOHOData.Resolutions.s512) extends
    SolarImagesSource {

    override def toString: String = "SOHO_"+instrument+"_"+size

  }

  case class SDO(
    override val instrument: String,
    override val size: Int = SDOData.Resolutions.s512) extends
    SolarImagesSource {

    override def toString: String = "SDO_"+instrument+"_"+size

  }

  case class GOES(
    quantity: String = GOESData.Quantities.XRAY_FLUX_5m,
    format: String = GOESData.Formats.CSV) extends Source

  case class OMNI(quantity: Int = OMNIData.Quantities.V_SW) extends Source {
    override def toString: String = "OMNI_"+OMNIData.columnNames(quantity).replace(" ", "_")
  }

  def image_process_metadata[S <: SolarImagesSource](image_source: S)
  : (DataPipe[Image, Image], Int, DataPipe[Image, Array[Byte]]) = image_source match {
    case _: SOHO => (
      DataPipe((i: Image) => i.filter(GrayscaleFilter)), 1,
      DataPipe((i: Image) => i.argb.map(_.last.toByte)))

    case SDO(AIA094335193, _) => (
      identityPipe[Image], 4,
      DataPipe((i: Image) => i.argb.flatten.map(_.toByte)))

    case SDO(HMI171, _) => (
      identityPipe[Image], 4,
      DataPipe((i: Image) => i.argb.flatten.map(_.toByte)))

    case _: SDO  => (
      DataPipe((i: Image) => i.filter(GrayscaleFilter)), 1,
      DataPipe((i: Image) => i.argb.map(_.last.toByte)))
  }

  private var size_buffer = 500

  private var image_byte_buffer = 40

  def buffer_size_(s: Int) = size_buffer = s

  def _buffer_size: Int = size_buffer

  def image_buffer_size_(s: Int) = image_byte_buffer = s

  def _image_buffer_size = image_byte_buffer

  val read_image = DataPipe((p: Path) => Image.fromPath(p.toNIO))

  def image_to_tensor = MetaPipe21[Int, Int, Array[Byte], Tensor](<commit_msg>[helios]: OMNI CTL codes   -- Possible fix to tensor from buffer bug: Buffered image loading approach<commit_after>package io.github.mandar2812.PlasmaML.helios

import ammonite.ops.{Path, pwd, root}
import breeze.linalg.DenseVector
import com.sksamuel.scrimage.Image
import com.sksamuel.scrimage.filter.GrayscaleFilter
import io.github.mandar2812.PlasmaML.helios
import io.github.mandar2812.PlasmaML.helios.data.SDOData.Instruments._
import io.github.mandar2812.PlasmaML.omni.{OMNIData, OMNILoader}
import io.github.mandar2812.dynaml.pipes._
import io.github.mandar2812.dynaml.analysis.implicits._
import io.github.mandar2812.dynaml.probability.{DiscreteDistrRV, MultinomialRV}
import io.github.mandar2812.dynaml.tensorflow.{dtf, dtfdata, dtfpipe}
import io.github.mandar2812.dynaml.tensorflow.data.{AbstractDataSet, DataSet, TFDataSet, ZipDataSet}
import io.github.mandar2812.dynaml.tensorflow.utils.{GaussianScalerTF, MinMaxScalerTF}
import io.github.mandar2812.dynaml.DynaMLPipe._
import io.github.mandar2812.dynaml.utils
import org.joda.time._
import spire.math.UByte
import org.platanios.tensorflow.api._


/**
  * <h3>Helios Data Facility</h3>
  *
  * Package enabling downloading and storage and processing
  * of heliospheric image data.
  *
  * @author mandar2812
  * */
package object data {


  private implicit val dateOrdering = new Ordering[DateTime] {
    override def compare(x: DateTime, y: DateTime): Int = if(x.isBefore(y)) -1 else 1
  }

  type IMAGE_PATTERN           = (DateTime, Path)

  /**
    * A simple data pattern, consisting of
    * a time stamp, path to an image, and a sequence of numbers
    * */
  type PATTERN                 = (DateTime, (Path, Seq[Double]))

  /**
    * A pattern, consisting of
    * a time stamp, path to an image, a tuple of numeric sequences
    * */
  type PATTERN_EXT             = (DateTime, (Path, (Seq[Double], Seq[Double])))

  /**
    * A pattern, consisting of
    * a time stamp, a collection of images from multiple sources,
    * and a sequence of numbers
    * */
  type MC_PATTERN              = (DateTime, (Map[SolarImagesSource, Seq[Path]], Seq[Double]))

  /**
    * A pattern, consisting of
    * a time stamp, a collection of images from multiple sources,
    * and a tuple of sequence of numbers
    * */
  type MC_PATTERN_EXT          = (DateTime, (Map[SOHO, Seq[Path]], (Seq[Double], Seq[Double])))

  type HELIOS_IMAGE_DATA       = DataSet[IMAGE_PATTERN]
  type HELIOS_OMNI_DATA        = DataSet[PATTERN]
  type HELIOS_MC_OMNI_DATA     = DataSet[MC_PATTERN]
  type HELIOS_OMNI_DATA_EXT    = DataSet[PATTERN_EXT]
  type HELIOS_MC_OMNI_DATA_EXT = Iterable[MC_PATTERN_EXT]

  type IMAGE_TS                = (Tensor, Tensor)

  type TF_IMAGE_DATA           = TFDataSet[Tensor]

  type TF_DATA                 = TFDataSet[(Tensor, Tensor)]

  type TF_DATA_EXT             = TFDataSet[((Tensor, Tensor), Tensor)]

  type TF_MC_DATA_EXT          = AbstractDataSet[(Tensor, Tensor), Tensor]

  type SC_TF_MC_DATA_EXT       = (TF_MC_DATA_EXT, (ReversibleScaler[(Tensor, Tensor)], MinMaxScalerTF))

  type SC_TF_DATA_EXT          = (TF_DATA_EXT, (ReversibleScaler[(Tensor, Tensor)], MinMaxScalerTF))

  type SC_TF_DATA              = (TF_DATA, (ReversibleScaler[Tensor], MinMaxScalerTF))

  private def TF_DATA_EXT(
    trData: IMAGE_TS,
    trLabels: Tensor,
    sizeTr: Int,
    tData: IMAGE_TS,
    tLabels: Tensor,
    sizeT: Int): AbstractDataSet[IMAGE_TS, Tensor] =
    AbstractDataSet(trData, trLabels, sizeTr, tData, tLabels, sizeT)



  /**
    * Download a resource (image, file) from a sequence of urls to a specified
    * disk location.
    * */
  def download_batch(path: Path)(urls: List[String]): Unit = {
    urls.par.foreach(s => utils.downloadURL(s, (path/s.split('/').last).toString()))
  }

  /**
    * Perform a bulk download of images within some date range
    * */
  def download_day_range(download: LocalDate => Unit)(start: LocalDate, end: LocalDate): Unit = {

    val num_days = new Duration(start.toDateTimeAtStartOfDay, end.toDateTimeAtStartOfDay).getStandardDays.toInt

    (0 to num_days).map(start.plusDays).par.foreach(download)
  }

  def download_month_range(download: YearMonth => Unit)(start: YearMonth, end: YearMonth): Unit = {

    val period = new Period(
      start.toLocalDate(1).toDateTimeAtStartOfDay,
      end.toLocalDate(31).toDateTimeAtStartOfDay)

    val num_months = (12*period.getYears) + period.getMonths

    (0 to num_months).map(start.plusMonths).par.foreach(download)
  }

  sealed trait Source
  sealed trait SolarImagesSource extends Source {
    val instrument: String
    val size: Int
  }

  case class SOHO(
    override val instrument: String,
    override val size: Int = SOHOData.Resolutions.s512) extends
    SolarImagesSource {

    override def toString: String = "SOHO_"+instrument+"_"+size

  }

  case class SDO(
    override val instrument: String,
    override val size: Int = SDOData.Resolutions.s512) extends
    SolarImagesSource {

    override def toString: String = "SDO_"+instrument+"_"+size

  }

  case class GOES(
    quantity: String = GOESData.Quantities.XRAY_FLUX_5m,
    format: String = GOESData.Formats.CSV) extends Source

  case class OMNI(quantity: Int = OMNIData.Quantities.V_SW) extends Source {
    override def toString: String = "OMNI_"+OMNIData.columnNames(quantity).replace(" ", "_")
  }

  def image_process_metadata[S <: SolarImagesSource](image_source: S)
  : (DataPipe[Image, Image], Int, DataPipe[Image, Array[Byte]]) = image_source match {
    case _: SOHO => (
      DataPipe((i: Image) => i.filter(GrayscaleFilter)), 1,
      DataPipe((i: Image) => i.argb.map(_.last.toByte)))

    case SDO(AIA094335193, _) => (
      identityPipe[Image], 4,
      DataPipe((i: Image) => i.argb.flatten.map(_.toByte)))

    case SDO(HMI171, _) => (
      identityPipe[Image], 4,
      DataPipe((i: Image) => i.argb.flatten.map(_.toByte)))

    case _: SDO  => (
      DataPipe((i: Image) => i.filter(GrayscaleFilter)), 1,
      DataPipe((i: Image) => i.argb.map(_.last.toByte)))
  }

  private var size_buffer = 500

  private var image_byte_buffer = 20

  def buffer_size_(s: Int) = size_buffer = s

  def _buffer_size: Int = size_buffer

  def image_buffer_size_(s: Int) = image_byte_buffer = s

  def _image_buffer_size = image_byte_buffer

  val read_image = DataPipe((p: Path) => Image.fromPath(p.toNIO))

  def image_to_tensor = MetaPipe21[Int, Int, Array[Byte], Tensor](
{'fraction_non_alphanumeric': 0.1140625, 'fraction_numerical': 0.01453125, 'mean_word_length': 3.6013659237958304, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6964544', 'n_tokens_mistral': 4581, 'n_tokens_neox': 4330, 'n_words': 985}