<commit_before>package it.polimi.genomics.spark.implementation.loaders

import java.io.InputStream

import it.polimi.genomics.core.ParsingType.PARSING_TYPE
import it.polimi.genomics.core.{GMQLSchemaCoordinateSystem, GMQLSchemaFormat, ParsingType, _}
import it.polimi.genomics.spark.utilities.FSConfig
import org.apache.hadoop.fs.{FileSystem, Path}
import org.slf4j.{Logger, LoggerFactory}

import scala.xml.XML

/**
  * Created by <NAME> on 25/05/15.
  */

/**
  * contains the helper functions for parsers
  */

object BedParserHelper {
  /**
    * returns parsed data as GValue by using parsingType and the value.
    *
    * @param parsingType
    * @param value
    * @return
    */
  def parseRegion(parsingType: ParsingType.Value, value: String): GValue = {
    parsingType match {
      case ParsingType.STRING | ParsingType.CHAR =>
        // TODO: better to differentiate char from string, however all the implementation does not take care of this
        GString(value.trim)
      case ParsingType.INTEGER | ParsingType.LONG | ParsingType.DOUBLE =>
        value.toLowerCase match {
          case "null" | "." =>
            GNull()
          case _ =>
            //No need to trim
            GDouble(value.toDouble)
        }
      case ParsingType.NULL =>
        GNull()
      case _ =>
        throw new Exception("Unknown ParsingType")
    }
  }


}


/**
  * GMQL Bed Parser, it is a parser that parse delimited text files,
  *
  * @param delimiter [[String]] of the delimiter used to separate columns, can be TAB, space, special Char, or something else.
  * @param chrPos    [[Int]] of the position index of chromosome column in the delimited file, this is compulsory column.
  * @param startPos  [[Int]] of the position index of region start column in the delimited file, this is compulsory column.
  * @param stopPos   [[Int]] of the position index of region stop column in the delimited file, this is compulsory column.
  * @param strandPos [[Int]] of the position index of strand column in the delimited file, this is compulsory column.
  * @param otherPos  [[Array]] of the other columns positions, this is [[Option]] and can be [[None]]. The Array has tuple of (position as [[Int]],[[ParsingType]])
  */
class BedParser(delimiter: String, var chrPos: Int, var startPos: Int, var stopPos: Int, var strandPos: Option[Int], var otherPos: Option[Array[(Int, ParsingType.PARSING_TYPE)]]) extends GMQLLoader[(Long, String), Option[DataTypes.GRECORD], (Long, String), Option[DataTypes.MetaType]] with java.io.Serializable {

  private val logger: Logger = LoggerFactory.getLogger(classOf[BedParser])
  var parsingType = GMQLSchemaFormat.TAB
  var coordinateSystem: GMQLSchemaCoordinateSystem.Value = GMQLSchemaCoordinateSystem.ZeroBased
  final val spaceDelimiter: String = " "
  final val semiCommaDelimiter: String = ";"
  private var otherPosGTF: Array[(Int, PARSING_TYPE)] = _
  private var namePositionMap: Map[String, Int] = _
  private var nameTypeMap: Map[String, PARSING_TYPE] = _

  //added function in order to calculate GTF parameters once.
  def calculateMapParameters(namePosition: Option[Seq[String]] = None): Unit = {
    parsingType match {
      case GMQLSchemaFormat.GTF =>
        if (otherPos.getOrElse(Array.empty).length > 4) otherPosGTF = otherPos.get.tail.tail.tail.tail else otherPosGTF = Array.empty
        //assume that namePosition order is same as otherPosGTF
        namePositionMap = namePosition.getOrElse(Iterable.empty).map(_.toUpperCase).zipWithIndex.toMap
        nameTypeMap = otherPosGTF.map(_._2).zip(namePosition.getOrElse(Seq.empty)).map(a => (a._2.toUpperCase, a._1)).toMap
    }
  }

  //  val otherPosExtended = otherPos.getOrElse(Array.empty).map()

  /**
    * Meta Data Parser to parse String to GMQL META TYPE (ATT, VALUE)
    *
    * @param t
    * @return
    */
  override def meta_parser(t: (Long, String)): Option[DataTypes.MetaType] = {
    val s = t._2.split(delimiter, -1)
    if (s.length != 2) {
      logger.warn("could not cast this line: \n\t\t" + t)
      None
    }
    else
      Some((t._1, (s(0), s(1))))
  }

  /**
    * Parser of String to GMQL Spark GRECORD
    *
    * @param t [[Tuple2]] of the ID and the [[String]] line to be parsed.
    * @return [[GRecord]] as GMQL record representation.
    */
  override def region_parser(t: (Long, String)): Option[DataTypes.GRECORD] = {
    import BedParserHelper._
    try {
      val s: Array[String] = t._2.split(delimiter, -1)

      val other = parsingType match {
        case GMQLSchemaFormat.GTF => {
          // GTF file format definition
          // 0) seqname - name of the chromosome or scaffold; chromosome names can be given with or without the 'chr' prefix. Important note: the seqname must be one used within Ensembl, i.e. a standard chromosome name or an Ensembl identifier such as a scaffold ID, without any additional content such as species or assembly. See the example GFF output below.
          // 1) source - name of the program that generated this feature, or the data source (database or project name)
          // 2) feature - feature type name, e.g. Gene, Variation, Similarity
          // 3) start - Start position of the feature, with sequence numbering starting at 1.
          // 4) end - End position of the feature, with sequence numbering starting at 1.
          // 5) score - A floating point value.
          // 6) strand - defined as + (forward) or - (reverse).
          // 7) frame - One of '0', '1' or '2'. '0' indicates that the first base of the feature is the first base of a codon, '1' that the second base is the first base of a codon, and so on..
          // 8) attribute - A semicolon-separated list of tag-value pairs, providing additional information about each feature.

          val source = parseRegion(ParsingType.STRING, s(1))
          val feature = parseRegion(ParsingType.STRING, s(2))
          val score = parseRegion(ParsingType.CHAR, s(5))
          val frame = parseRegion(ParsingType.STRING, s(7))


          val restValues = Array.fill[GValue](otherPosGTF.length)(GNull())

          val values = s(8) split semiCommaDelimiter<commit_msg>Fixes an error in the GTF parser (score was parsed as a CHAR instead of DOUBLE)<commit_after>package it.polimi.genomics.spark.implementation.loaders

import java.io.InputStream

import it.polimi.genomics.core.ParsingType.PARSING_TYPE
import it.polimi.genomics.core.{GMQLSchemaCoordinateSystem, GMQLSchemaFormat, ParsingType, _}
import it.polimi.genomics.spark.utilities.FSConfig
import org.apache.hadoop.fs.{FileSystem, Path}
import org.slf4j.{Logger, LoggerFactory}

import scala.xml.XML

/**
  * Created by <NAME> on 25/05/15.
  */

/**
  * contains the helper functions for parsers
  */

object BedParserHelper {
  /**
    * returns parsed data as GValue by using parsingType and the value.
    *
    * @param parsingType
    * @param value
    * @return
    */
  def parseRegion(parsingType: ParsingType.Value, value: String): GValue = {
    parsingType match {
      case ParsingType.STRING | ParsingType.CHAR =>
        // TODO: better to differentiate char from string, however all the implementation does not take care of this
        GString(value.trim)
      case ParsingType.INTEGER | ParsingType.LONG | ParsingType.DOUBLE =>
        value.toLowerCase match {
          case "null" | "." =>
            GNull()
          case _ =>
            //No need to trim
            GDouble(value.toDouble)
        }
      case ParsingType.NULL =>
        GNull()
      case _ =>
        throw new Exception("Unknown ParsingType")
    }
  }


}


/**
  * GMQL Bed Parser, it is a parser that parse delimited text files,
  *
  * @param delimiter [[String]] of the delimiter used to separate columns, can be TAB, space, special Char, or something else.
  * @param chrPos    [[Int]] of the position index of chromosome column in the delimited file, this is compulsory column.
  * @param startPos  [[Int]] of the position index of region start column in the delimited file, this is compulsory column.
  * @param stopPos   [[Int]] of the position index of region stop column in the delimited file, this is compulsory column.
  * @param strandPos [[Int]] of the position index of strand column in the delimited file, this is compulsory column.
  * @param otherPos  [[Array]] of the other columns positions, this is [[Option]] and can be [[None]]. The Array has tuple of (position as [[Int]],[[ParsingType]])
  */
class BedParser(delimiter: String, var chrPos: Int, var startPos: Int, var stopPos: Int, var strandPos: Option[Int], var otherPos: Option[Array[(Int, ParsingType.PARSING_TYPE)]]) extends GMQLLoader[(Long, String), Option[DataTypes.GRECORD], (Long, String), Option[DataTypes.MetaType]] with java.io.Serializable {

  private val logger: Logger = LoggerFactory.getLogger(classOf[BedParser])
  var parsingType = GMQLSchemaFormat.TAB
  var coordinateSystem: GMQLSchemaCoordinateSystem.Value = GMQLSchemaCoordinateSystem.ZeroBased
  final val spaceDelimiter: String = " "
  final val semiCommaDelimiter: String = ";"
  private var otherPosGTF: Array[(Int, PARSING_TYPE)] = _
  private var namePositionMap: Map[String, Int] = _
  private var nameTypeMap: Map[String, PARSING_TYPE] = _

  //added function in order to calculate GTF parameters once.
  def calculateMapParameters(namePosition: Option[Seq[String]] = None): Unit = {
    parsingType match {
      case GMQLSchemaFormat.GTF =>
        if (otherPos.getOrElse(Array.empty).length > 4) otherPosGTF = otherPos.get.tail.tail.tail.tail else otherPosGTF = Array.empty
        //assume that namePosition order is same as otherPosGTF
        namePositionMap = namePosition.getOrElse(Iterable.empty).map(_.toUpperCase).zipWithIndex.toMap
        nameTypeMap = otherPosGTF.map(_._2).zip(namePosition.getOrElse(Seq.empty)).map(a => (a._2.toUpperCase, a._1)).toMap
    }
  }

  //  val otherPosExtended = otherPos.getOrElse(Array.empty).map()

  /**
    * Meta Data Parser to parse String to GMQL META TYPE (ATT, VALUE)
    *
    * @param t
    * @return
    */
  override def meta_parser(t: (Long, String)): Option[DataTypes.MetaType] = {
    val s = t._2.split(delimiter, -1)
    if (s.length != 2) {
      logger.warn("could not cast this line: \n\t\t" + t)
      None
    }
    else
      Some((t._1, (s(0), s(1))))
  }

  /**
    * Parser of String to GMQL Spark GRECORD
    *
    * @param t [[Tuple2]] of the ID and the [[String]] line to be parsed.
    * @return [[GRecord]] as GMQL record representation.
    */
  override def region_parser(t: (Long, String)): Option[DataTypes.GRECORD] = {
    import BedParserHelper._
    try {
      val s: Array[String] = t._2.split(delimiter, -1)

      val other = parsingType match {
        case GMQLSchemaFormat.GTF => {
          // GTF file format definition
          // 0) seqname - name of the chromosome or scaffold; chromosome names can be given with or without the 'chr' prefix. Important note: the seqname must be one used within Ensembl, i.e. a standard chromosome name or an Ensembl identifier such as a scaffold ID, without any additional content such as species or assembly. See the example GFF output below.
          // 1) source - name of the program that generated this feature, or the data source (database or project name)
          // 2) feature - feature type name, e.g. Gene, Variation, Similarity
          // 3) start - Start position of the feature, with sequence numbering starting at 1.
          // 4) end - End position of the feature, with sequence numbering starting at 1.
          // 5) score - A floating point value.
          // 6) strand - defined as + (forward) or - (reverse).
          // 7) frame - One of '0', '1' or '2'. '0' indicates that the first base of the feature is the first base of a codon, '1' that the second base is the first base of a codon, and so on..
          // 8) attribute - A semicolon-separated list of tag-value pairs, providing additional information about each feature.

          val source = parseRegion(ParsingType.STRING, s(1))
          val feature = parseRegion(ParsingType.STRING, s(2))
          val score = parseRegion(ParsingType.DOUBLE, s(5))
          val frame = parseRegion(ParsingType.STRING, s(7))


          val restValues = Array.fill[GValue](otherPosGTF.length)(GNull())

          val values = s(8) split semiCommaDelimiter
{'fraction_non_alphanumeric': 0.102460343153124, 'fraction_numerical': 0.0066364519261897055, 'mean_word_length': 3.094433399602386, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '1570315', 'n_tokens_mistral': 3750, 'n_tokens_neox': 3598, 'n_words': 1266}