process_data.sc

import scala.concurrent.{Await, Future}
import os.Path
import scala.util.NotGiven
import scala.deriving.Mirror
import scala.compiletime.{summonAll}
import scala.util.Using
import scala.reflect.ClassTag
import java.time.Duration
import scala.util.chaining.given

trait CSVConvert[T]:
  def toString(value: T): String
  def fromString(str: String): T

object CSVConvert:
  def apply[T: CSVConvert](value: T): String =
    summon[CSVConvert[T]].toString(value)

  given stringConv: CSVConvert[String] with
    def fromString(str: String): String = str
    def toString(value: String): String = value

  given boolConv: CSVConvert[Boolean] with
    def fromString(str: String): Boolean =
      str match
        case "true"  => true
        case "false" => false
    def toString(value: Boolean): String =
      if value then "true" else "false"

  given intConv: CSVConvert[Int] with
    def fromString(str: String): Int = str.toInt
    def toString(value: Int): String = value.toString

  given longConv: CSVConvert[Long] with
    def fromString(str: String): Long = str.toLong
    def toString(value: Long): String = value.toString

  given doubleConv: CSVConvert[Double] with
    def fromString(str: String): Double = str.toDouble
    def toString(value: Double): String = value.toString

  given pathConv: CSVConvert[os.Path] with
    def fromString(str: String): Path = os.Path(str, os.pwd)
    def toString(value: Path): String = value.relativeTo(os.pwd).toString

  given durationConv: CSVConvert[Duration] with
    def fromString(str: String): Duration =
      val Array(seconds, nanos) = str.split("\\+"): @unchecked
      Duration.ZERO
        .plus(Duration.ofSeconds(seconds.toLong))
        .plus(Duration.ofNanos(nanos.toLong))
    def toString(value: Duration): String =
      s"${value.getSeconds()}+${value.getNano()}"

trait CSVRowConvert[Row]:
  def rowLength: Int
  def toRow(row: Row): List[String]
  def fromRow(row: List[String]): Row

object CSVRowConvert:
  given emptyCase: CSVRowConvert[EmptyTuple] with
    def rowLength: Int = 0
    def toRow(row: EmptyTuple): List[String] = Nil
    def fromRow(row: List[String]): EmptyTuple =
      require(row.isEmpty)
      EmptyTuple

  given consCase[Elem, Row <: Tuple](using
      elemConv: CSVConvert[Elem],
      restConv: CSVRowConvert[Row]
  ): CSVRowConvert[Elem *: Row] with
    def rowLength: Int = restConv.rowLength + 1
    def toRow(row: Elem *: Row): List[String] =
      elemConv.toString(row.head) :: restConv.toRow(row.tail)
    def fromRow(row: List[String]): Elem *: Row =
      elemConv.fromString(row.head) *: restConv.fromRow(row.tail)

  given productCase[T <: Product](using
      mirror: Mirror.ProductOf[T]
  )(using
      NotGiven[T <:< Tuple]
  )(using conv: CSVRowConvert[mirror.MirroredElemTypes]): CSVRowConvert[T] with
    def rowLength: Int = conv.rowLength
    def toRow(row: T): List[String] = conv.toRow(Tuple.fromProductTyped(row))
    def fromRow(row: List[String]): T = mirror.fromProduct(conv.fromRow(row))

extension [Row](self: Row)
  def toCSVRow(using rowConvert: CSVRowConvert[Row]): List[String] =
    rowConvert.toRow(self)

extension [Row](iter: IterableOnce[Row])
  def toCSV(outPath: os.Path)(headings: List[String])(using
      NotGiven[Row <:< String]
  )(using rowConvert: CSVRowConvert[Row]): Unit =
    require(
      headings.size == rowConvert.rowLength,
      s"headings did not match row length ($headings)"
    )
    Using.resource(com.github.tototoshi.csv.CSVWriter.open(outPath.toIO)):
      writer =>
        writer.writeRow(headings)
        iter.iterator.foreach: row =>
          writer.writeRow(row.toCSVRow)

extension (iter: IterableOnce[List[String]])
  def toCSV(outPath: os.Path)(headings: List[String]): Unit =
    Using.resource(com.github.tototoshi.csv.CSVWriter.open(outPath.toIO)):
      writer =>
        writer.writeRow(headings)
        iter.iterator.foreach: row =>
          assert(
            row.size == headings.size,
            s"row of size ${row.size} does not match heading count ${headings.size}. row = $row; headings = $headings"
          )
          writer.writeRow(row)

extension (fileName: os.Path)
  def fromCSV[Row](headings: List[String])(using
      rowConvert: CSVRowConvert[Row]
  ): List[Row] =
    require(
      headings.size == rowConvert.rowLength,
      "headings did not match row length"
    )
    Using.resource(com.github.tototoshi.csv.CSVReader.open(fileName.toIO)):
      reader =>
        val it = reader.iterator
        assert(headings == it.next(), "headings did not match")
        it
          .map(_.toList)
          .map(rowConvert.fromRow)
          .toList

val checkFiles = os
  .walk(os.pwd / "systems")
  .filter(os.isFile)
  .filter(_.last.endsWith("_check.txt"))

extension (duration: Duration)
  def toDouble: Double =
    duration.getSeconds().toDouble + duration.getNano().toDouble / 1_000_000_000

def toDuration(double: Double): Duration =
  val seconds = double.toLong
  Duration.ZERO
    .plus(Duration.ofNanos((double - seconds * 1_000_000_000).toLong))
    .plus(Duration.ofSeconds(seconds))

val ElapsedTime = raw""".*system (\d+):(\d+).(\d+)elapsed.*""".r
val GraphDepth =
  raw"""The depth of the complete state graph search is (\d+)\.""".r
val DistinctStates =
  raw"""\d+ states generated, (\d+) distinct states found, \d+ states left on queue\.""".r
val ViolatedByInitialState =
  raw"""Error: Invariant \w+ is violated by the initial state:""".r

final case class DataPoint(
    fileName: os.Path,
    id: String,
    specName: String,
    traceVariant: String,
    src: Int,
    success: Boolean,
    isDFS: Boolean,
    isProgressInv: Boolean,
    genRuntime: Duration,
    checkRuntime: Duration,
    graphDepth: Long,
    distinctStates: Long,
    checkOutputSize: Long,
    logLineCount: Long
):
  def expectedBugs: Set[String] =
    var result = Set.empty[String]

    if traceVariant.contains("bad_vclocks")
    then result += "bugshared"
    if traceVariant.contains("netLen")
    then result += "buglenimpl"

    if specName.contains("_tcp")
    then result += "bugtcp"
    if specName.contains("_netLen")
    then result += "bugunderbuf"
    if specName.contains("_clientTimeout")
    then result += "bugtimeoutclient"
    if specName.contains("_leaderTimeout")
    then result += "bugtimeoutleader"
    if specName.contains("_fd")
    then result += "bugfd"

    if specName.contains("miscomp")
    then result += "bugrequests"

    result
  end expectedBugs

  def specNameInvar: String =
    specName
      .stripSuffix("_tcp")
      .stripSuffix("_netLen")
      .stripSuffix("_clientTimeout")
      .stripSuffix("_leaderTimeout")
      .stripSuffix("_miscomp")
      .stripSuffix("_fd")
end DataPoint

object DataPoint:
  inline def csvHeader(using mirror: Mirror.ProductOf[DataPoint]): Seq[String] =
    given [T](using value: ValueOf[T]): T = value.value
    summonAll[mirror.MirroredElemLabels].toArray
      .map(_.asInstanceOf[String])
      .toSeq

  def apply(fileName: os.Path): DataPoint =
    val id = (fileName / os.up).last
    val src: Int =
      if fileName.last.startsWith("2_")
      then 2
      else if fileName.last.startsWith("3_")
      then 3
      else 1

    val rootName: String =
      fileName.last
        .stripPrefix("2_")
        .stripPrefix("3_")
        .stripSuffix("_check.txt")

    def lines =
      os.read.lines.stream(fileName)

    def findUnique[T: ClassTag](gen: os.Generator[T]): T =
      val values = gen.toArray
      require(
        values.size == 1,
        s"number of instances ${values.size} was not 1 in $fileName"
      )
      values.head

    def findOptional[T: ClassTag](gen: os.Generator[T]): Option[T] =
      val values = gen.toArray
      require(
        values.size <= 1,
        s"number of instances ${values.size} was greater than 1 in $fileName"
      )
      values.headOption

    def uniqueLine[T: ClassTag](fn: PartialFunction[String, T]): T =
      findUnique(lines.collect(fn))

    def optionalLine[T: ClassTag](fn: PartialFunction[String, T]): Option[T] =
      findOptional(lines.collect(fn))

    val success: Boolean =
      lines.exists(
        _.contains("Model checking completed. No error has been found.")
      )

    val checkRuntime: Duration =
      uniqueLine:
        case ElapsedTime(minutes, seconds, hundredths) =>
          Duration.ZERO
            .plus(Duration.ofMinutes(minutes.toLong))
            .plus(Duration.ofSeconds(seconds.toLong))
            .plus(Duration.ofMillis(hundredths.toLong * 10))

    val graphDepth: Long =
      optionalLine:
        case ViolatedByInitialState() => 1L
      .getOrElse:
        uniqueLine:
          case GraphDepth(depth) => depth.toLong

    val distinctStates: Long =
      optionalLine:
        case ViolatedByInitialState() => 1L
      .getOrElse:
        uniqueLine:
          case DistinctStates(states) => states.toLong

    val genRuntime: Duration =
      findUnique:
        os.read.lines
          .stream(
            fileName / os.up / s"${fileName.last.stripSuffix("_check.txt")}_gen.txt"
          )
          .collect:
            case ElapsedTime(minutes, seconds, hundredths) =>
              Duration.ZERO
                .plus(Duration.ofMinutes(minutes.toLong))
                .plus(Duration.ofSeconds(seconds.toLong))
                .plus(Duration.ofMillis(hundredths.toLong * 10))

    val checkOutputSize: Long =
      os.size(fileName)

    val isDFS: Boolean =
      rootName.contains("_dfs")

    val isProgressInv: Boolean =
      !rootName.contains("_noprogressinv")

    val specName: String =
      rootName
        .stripSuffix("_check")
        .stripSuffix("_noprogressinv")
        .stripSuffix("_dfs")

    val logLineCount: Long =
      os.list(fileName / os.up)
        .filter(_.last.endsWith(".log"))
        .map: logFile =>
          os.read.lines.stream(logFile).count(_ => true)
        .sum

    val traceVariant: String =
      (fileName / os.up / os.up).last
        .stripPrefix("traces_found")
        .stripPrefix("_")

    DataPoint(
      fileName = fileName,
      id = id,
      specName = specName,
      traceVariant = traceVariant,
      src = src,
      success = success,
      isDFS = isDFS,
      isProgressInv = isProgressInv,
      genRuntime = genRuntime,
      checkRuntime = checkRuntime,
      graphDepth = graphDepth,
      distinctStates = distinctStates,
      checkOutputSize = checkOutputSize,
      logLineCount = logLineCount
    )

val dataPoints =
  given scala.concurrent.ExecutionContext =
    scala.concurrent.ExecutionContext.global

  val csvName = os.pwd / "all_data.csv"
  val csvHeader: List[String] = DataPoint.csvHeader.toList
  val origPoints: List[DataPoint] =
    if os.exists(csvName)
    then csvName.fromCSV[DataPoint](csvHeader)
    else Nil
  val byProcessedFileName =
    origPoints.iterator.map(pt => pt.fileName -> pt).toMap

  Using
    .Manager: use =>
      val writer = use(com.github.tototoshi.csv.CSVWriter.open(csvName.toIO))
      writer.writeRow(csvHeader)

      checkFiles.iterator
        .map: fileName =>
          byProcessedFileName.get(fileName) match
            case None =>
              Future:
                // println(s"read $fileName")
                DataPoint(fileName)
                  .tap(pt => writer.synchronized(writer.writeRow(pt.toCSVRow)))
            case Some(pt) =>
              // println(s"skip $fileName")
              writer.synchronized(writer.writeRow(pt.toCSVRow))
              Future.successful(pt)
        .foldRight(Future(Nil: List[DataPoint])): (mkPt, accRest) =>
          for
            pt <- mkPt
            rest <- accRest
          yield pt :: rest
        .pipe(Await.result(_, scala.concurrent.duration.Duration.Inf))
    .get
end dataPoints

// For each spec validation, report data per traversal (pclock, bfs, dfs, sidestep), report: times, depths, trace size, etc

// For each spec / bug combo, report false positive rate for each traversal (pclock, bfs, dfs, sidestep), double column with % and pos/neg metrics

final case class Reps(reps: List[DataPoint]):
  def propUniq[T](accessor: DataPoint => T): T =
    val values = reps.iterator.map(accessor).toSet
    assert(values.size == 1)
    values.head

  def propAvg[T: Numeric](accessor: DataPoint => T): Double =
    val values = reps.map(accessor)
    summon[Numeric[T]].toDouble(values.sum) / values.size

  def specName: String = propUniq(_.specName)
  def traceVariant: String = propUniq(_.traceVariant)
  def isDFS: Boolean = propUniq(_.isDFS)
  def isProgressInv: Boolean = propUniq(_.isProgressInv)
  def success: Boolean = propUniq(_.success)
  def expectedBugs: Set[String] = propUniq(_.expectedBugs)
  def fileName: os.Path = propUniq(pt =>
    pt.fileName / os.up / pt.fileName.last.stripPrefix("2_").stripPrefix("3_")
  )

  def checkRuntime: Double = propAvg(_.checkRuntime.toDouble)
  def graphDepth: Double = propAvg(_.graphDepth.toDouble)

val dataPointsByRepetition = dataPoints
  .groupBy(pt =>
    pt.fileName / os.up / pt.fileName.last.stripPrefix("2_").stripPrefix("3_")
  )
  .iterator
  .map(_._2)
  .map(Reps.apply)
  .toList

dataPointsByRepetition
  .filter(_.specName.contains("raftkvs"))
  .groupBy(pt => (pt.specName, pt.traceVariant))
  .foreach: (pair, pts) =>
    if pair == ("raftkvs", "5")
    then assert(pts.size == 19) // we lost one due to Ctrl^c
    else
      assert(
        pts.size == 10 || pts.size == 20,
        s"for $pair, had ${pts.size} records instead of 10 or 20"
      )

val specMetrics =
  dataPointsByRepetition
    .filter(_.expectedBugs.isEmpty)
    .groupBy(pt => (pt.specName, pt.traceVariant))
    .iterator
    .map:
      case ((specName, traceVariant), pts) =>
        val bfsOnlyRecs = pts.filter(pt => !pt.isDFS && !pt.isProgressInv)
        val dfsOnlyRecs = pts.filter(pt => pt.isDFS && !pt.isProgressInv)
        val bfsProgressInvRecs = pts.filter(pt => !pt.isDFS && pt.isProgressInv)
        val dfsProgressInvRecs = pts.filter(pt => pt.isDFS && pt.isProgressInv)

        def avgCheckTime(reps: List[Reps]): Double =
          reps.map(_.checkRuntime).sum / reps.size

        val displayName =
          if traceVariant != ""
          then s"$specName, $traceVariant"
          else specName

        (
          displayName,
          if bfsOnlyRecs.isEmpty then ""
          else f"${avgCheckTime(bfsOnlyRecs)}%.2f",
          if bfsProgressInvRecs.isEmpty then ""
          else f"${avgCheckTime(bfsProgressInvRecs)}%.2f",
          f"${avgCheckTime(dfsOnlyRecs)}%.2f",
          f"${avgCheckTime(dfsProgressInvRecs)}%.2f"
        )
    .toCSV(os.pwd / "validation_metrics.csv")(
      List(
        "config",
        "time (bfs)",
        "time (bfs + sidestep)",
        "time (dfs)",
        "time (dfs + sidestep)"
      )
    )
end specMetrics

val bugMetrics =
  val allBugs =
    dataPoints.iterator
      .flatMap(_.expectedBugs)
      .toArray
      .sortInPlace()
      .distinct

  dataPointsByRepetition
    .filter(_.expectedBugs.sizeIs == 1)
    .filter(_.isDFS)
    // .filter(_.specName.contains("raftkvs"))
    .groupBy: pt =>
      val shortSpec =
        if pt.specName.contains("_")
        then pt.specName.split("_").head
        else pt.specName
      val shortVariant =
        if pt.traceVariant == "bad_vclocks"
        then ""
        else if pt.traceVariant != "3_fail" && pt.traceVariant.contains("_")
        then pt.traceVariant.split("_").head
        else pt.traceVariant
      (shortSpec, shortVariant)
    .iterator
    .map:
      case ((specName, traceVariant), pts) =>
        val displayName =
          if traceVariant == "3_fail"
          then s"$specName, n = 3 +fail"
          else if traceVariant != ""
          then s"$specName, n = $traceVariant"
          else specName

        val ptsDfs = pts.filter(!_.isProgressInv)
        val ptsSidestep = pts.filter(_.isProgressInv)
        assert(ptsDfs.size == ptsSidestep.size)

        val bugsFoundDfs = ptsDfs.count(!_.success).toDouble / ptsDfs.size * 100
        val findTimeAvgDfs = ptsDfs
          .filter(!_.success)
          .pipe(pts => pts.map(_.checkRuntime).sum / pts.size)
        val graphDepthAvgDfs = ptsDfs
          .filter(!_.success)
          .pipe(pts => pts.map(_.graphDepth.toDouble).sum / pts.size)
        val bugsFoundSidestep = ptsSidestep
          .filter(_.isProgressInv)
          .count(!_.success)
          .toDouble / ptsSidestep.size * 100
        val findTimeAvgSidestep = ptsSidestep
          .filter(!_.success)
          .pipe(pts => pts.map(_.checkRuntime).sum / pts.size)
        val graphDepthAvgSidestep = ptsSidestep
          .filter(!_.success)
          .pipe(pts => pts.map(_.graphDepth.toDouble).sum / pts.size)

        (
          displayName,
          ptsDfs.size,
          f"$bugsFoundDfs%.2f%%",
          f"$findTimeAvgDfs%.1f",
          f"$graphDepthAvgDfs%.0f",
          f"$bugsFoundSidestep%.2f%%",
          f"$findTimeAvgSidestep%.1f",
          f"$graphDepthAvgSidestep%.0f"
        )
    .toCSV(os.pwd / "bug_metrics.csv")(
      List(
        "config",
        "validation count",
        "bugs found",
        "find time",
        "depth",
        "bugs found (sidestep)",
        "find time (sidestep)",
        "depth (sidestep)"
      )
    )
end bugMetrics