//设置package包名称以及导入依赖的类
package com.newegg.eims.DataPorter.Csv

import java.io.{File, FileWriter}

import com.newegg.eims.DataPorter.Base.DataSet
import org.apache.commons.csv.{CSVFormat, CSVPrinter}

import scala.collection.JavaConverters._


    def saveCsv(path: String, isAppend: Boolean = false, format: CSVFormat = null): File = {
      val csvFormat = if (format != null) format
      else CSVFormat.newFormat(';')
        .withTrim(true)
        .withRecordSeparator("\r\n")
        .withIgnoreEmptyLines()
        .withIgnoreHeaderCase()
      val rows = set.toRowIterator
      val cols = rows.getSchema.getColumns
      val writer = new FileWriter(path, isAppend)
      val printer = new CSVPrinter(writer, csvFormat)
      try {
        rows.map(r => cols.map(c => c.getValFromRow(r).getOrElse("").toString).toSeq.asJavaCollection)
          .foreach(i => printer.printRecord(i))
        writer.flush()
      } finally {
        writer.close()
        printer.close()
      }
      new File(path)
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.newegg.eims.DataPorter.Csv

import java.io.File

import com.newegg.eims.DataPorter.Base.Converts._
import com.newegg.eims.DataPorter.Base._
import com.newegg.eims.DataPorter.Csv.Converts._
import org.apache.commons.csv.CSVFormat
import org.scalatest.{FlatSpec, Matchers}

import scala.annotation.meta.field
import scala.io.Source



case class TestD1(@([email protected])(CInt) a: Int, @([email protected])(CString) b: String)

class CsvDataSetWriterSpec extends FlatSpec with Matchers {
  val data = Array(1, 2).toIterable.transform(
    new DataColumn(0, "a", CInt, i => Some(i * 5)),
    new DataColumn(1, "b", CString, i => Some(i.toString))
  )
  val currentDir = new File(".").getCanonicalPath + File.separator + "target" + File.separator

  "CscDataSet" should "saveCsv with no format and no append" in {
    val f = data.saveCsv(currentDir + "test.csv")
    f.exists() shouldBe true
    val source = Source.fromFile(f)
    val info = try source.mkString finally source.close()
    info shouldBe "5;1\r\n10;2\r\n"
    f.delete() shouldBe true
  }

  it should "saveCsv with custom format and append" in {
    val format = CSVFormat.newFormat(',').withTrim(true).withRecordSeparator("\r\n")
    data.saveCsv(currentDir + "test1.csv", isAppend = false, format).exists() shouldBe true
    val f = data.as[TestD1]().saveCsv(currentDir + "test1.csv", isAppend = true, format)
    f.exists() shouldBe true
    val source = Source.fromFile(f)
    val info = try source.mkString finally source.close()
    info shouldBe "5,1\r\n10,2\r\n5,1\r\n10,2\r\n"
  }
} 

//设置package包名称以及导入依赖的类
package com.cloudera.datascience.cdsw.acme

import java.nio.charset.StandardCharsets
import java.nio.file.{Files, Paths}
import java.nio.file.StandardOpenOption._
import scala.collection.JavaConverters._

import org.apache.commons.csv.CSVFormat
import org.dmg.pmml.FieldName
import org.jpmml.evaluator.{ModelEvaluatorFactory, ProbabilityDistribution}
import org.jpmml.model.PMMLUtil
import org.scalatest.{FlatSpec, Matchers}

class PMMLModelSuite extends FlatSpec with Matchers {
  
  "model" should "be 90+% accurate" in {
    val modelPath = Paths.get("src", "main", "resources", "model.pmml")
    val stream = Files.newInputStream(modelPath, READ)
    val pmml = 
      try {
        PMMLUtil.unmarshal(stream)
      } finally {
        stream.close()
      }
    
    val evaluator = ModelEvaluatorFactory.newInstance().newModelEvaluator(pmml)
    evaluator.verify()
    
    var correct = 0
    var total = 0
    
    val testPath = Paths.get("src", "test", "resources", "datatest.csv")
    val testReader = Files.newBufferedReader(testPath, StandardCharsets.UTF_8)
    try {
      CSVFormat.RFC4180.withFirstRecordAsHeader().parse(testReader).asScala.foreach { record =>
        val inputMap = record.toMap.asScala.
          filterKeys(_ != "Occupancy").
          map { case (field, fieldValue) => (new FieldName(field), fieldValue) }.asJava
        val outputMap = evaluator.evaluate(inputMap)
        
        val expected = record.get("Occupancy").toInt
        val actual = outputMap.get(new FieldName("Occupancy")).
          asInstanceOf[ProbabilityDistribution].getResult.toString.toInt
        
        if (expected == actual) {
          correct += 1
        }
        total += 1
      }
    } finally {
      testReader.close()
    }
    
    val accuracy = correct.toDouble / total
    println(s"Accuracy: $accuracy")
    assert(accuracy >= 0.9)
  }

} 

//设置package包名称以及导入依赖的类
package teleporter.integration.component

import java.io.{InputStream, InputStreamReader}
import java.util

import akka.NotUsed
import akka.stream._
import akka.stream.scaladsl.Flow
import akka.stream.stage.{GraphStage, GraphStageLogic, InHandler, OutHandler}
import akka.util.ByteString
import org.apache.commons.csv.{CSVFormat, CSVParser, CSVRecord}


class CSVStage(format: CSVFormat, val maximumLineBytes: Int) extends GraphStage[FlowShape[ByteString, CSVRecord]] {
  val in: Inlet[ByteString] = Inlet[ByteString]("CsvStage.in")
  val out: Outlet[CSVRecord] = Outlet[CSVRecord]("CsvStage.out")
  override val shape: FlowShape[ByteString, CSVRecord] = FlowShape(in, out)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler {
    private var buffer = ByteString.empty
    private var cursor = -1
    private val reader = new InputStreamReader(new InputStream {
      override def read(): Int = {
        cursor += 1
        require(cursor < buffer.length, s"cursor must little than ${buffer.length}, Please increase $maximumLineBytes")
        buffer(cursor)
      }
    })
    private val csvRecords: util.Iterator[CSVRecord] = new CSVParser(reader, format).iterator()

    override def onPush(): Unit = {
      buffer ++= grab(in)
      if (buffer.length < maximumLineBytes) {
        pull(in)
      } else {
        if (csvRecords.hasNext) {
          push(out, csvRecords.next())
          buffer = buffer.drop(cursor).compact
          cursor = -1
        }
      }
    }

    override def onPull(): Unit = {
      pull(in)
    }

    override def onUpstreamFinish(): Unit = {
      if (buffer.isEmpty) {
        completeStage()
      } else if (isAvailable(out)) {
        while (csvRecords.hasNext) {
          emit(out, csvRecords.next())
        }
        completeStage()
      }
    }

    setHandlers(in, out, this)
  }
} 

//设置package包名称以及导入依赖的类
package teleporter.integration.component

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import akka.util.ByteString
import org.apache.commons.csv.CSVFormat
import org.scalatest.FunSuite

import scala.concurrent.Await
import scala.concurrent.duration._


class CSV$Test extends FunSuite {
  implicit val system = ActorSystem()
  implicit val mater = ActorMaterializer()
  val text: String =
    """
      |aaaaa,"bb
      |b,
      |bb",ccccc!
      |aaaa1,bbbbb,ccccc!
      |aaaa2,bbbbb,ccccc!
      |aaaa3,bbbbb,ccccc!
      |aaaa4,bbbbb,ccccc!
      |aaaa5,bbbbb,ccccc!
    """.stripMargin
  test("csv test") {
    val fu = Source(text.grouped(10).toIndexedSeq).map(ByteString(_))
      .via(CSV.flow(CSVFormat.DEFAULT.withRecordSeparator('\n'), 5))
      .runForeach(println(_))
    Await.result(fu, 1.minute)
  }
}