// ---
// jupyter:
//   jupytext:
//     text_representation:
//       extension: .scala
//       format_name: light
//       format_version: '1.5'
//       jupytext_version: 1.14.4
//   kernelspec:
//     display_name: Scala
//     language: scala
//     name: scala
// ---

// ## Evidence Preparation Error Summary Report

import $ivy.`org.plotly-scala::plotly-almond:0.7.2`
import $file.^.sparkinit, sparkinit._
import $file.^.pathinit, pathinit._
import $file.^.cpinit, cpinit._
import ss.implicits._
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import java.nio.file.Paths
import plotly._
import plotly.element._
import plotly.layout._
import plotly.Almond._
implicit class DFOPs(df: DataFrame) { def fn[T](fn: DataFrame => T): T = fn(df)}

init(offline=false)

// ### Load Summary Datasets

val regex = ".*evidence_(.*)_validation_.*".r
val steps = RESULTS_DIR.resolve("errors").toFile.listFiles.map(_.toString).map({
    case regex(k) => k
    case _ => ""
}).toSet
def loadSummary(step: String) = ss.read
    .parquet(RESULTS_DIR.resolve(s"errors/evidence_${step}_validation_summary.parquet").toString)
    .withColumn("step", lit(step))
val df = steps.map(loadSummary).reduce(_.union(_))

df.withColumn("reason", coalesce($"reason", lit("none"))).show(1000, false)

// ### Record Invalidation Cause Frequency by Source
//
// Evidence records are eliminated in the pipeline for a variety of reasons, and this section shows the the frequency with which those conditions are encountered per data source.  Reason = "none" below indicates that the record passed all validation filters (i.e. only these records are kept, all others are lost).

// Visualize the number of records with some reason they were invalid alongside those that were not (reason = "none")
val traces = df.select("step").dropDuplicates().collect().map(_.getAs[String]("step")).toList.map(s => 
        df
            .withColumn("reason", coalesce($"reason", lit("none")))
            .withColumn("sourceID", coalesce($"sourceID", lit("none")))
            .filter($"step" === s)
            .fn(ds => {
                s -> ds.select("reason").dropDuplicates().collect.map(_.getAs[String]("reason")).toSeq.map(r => {
                    ds
                        .filter($"reason" === r)
                        .sort($"count".desc)
                        .fn( dsr => {
                            Bar(
                                x=dsr.select("sourceID").collect.map(_.getAs[String]("sourceID")).toList,
                                y=dsr.select("count").collect.map(_.getAs[Long]("count")).toList,
                                name=r,
                                showlegend=true
                            )
                        })
                })
            })
    ).toMap

traces.foreach {case (k, data) => {
    data.plot(
        title=s"Validation phase: $k", 
        yaxis=Axis(`type`=AxisType.Log),
        margin=Margin(t=40),
        barmode=BarMode.Group
    )
}}

// Show only non-valid record counts for reference:

df.filter($"reason".isNotNull).show(100, false)

StarcoderdataJupyterScriptsDedupFiltered