<commit_before>package intelmedia.ws.monitoring

import java.util.concurrent.ExecutorService
import scala.concurrent.duration._
import scalaz.concurrent.{Strategy,Task}
import scalaz.stream._
import pru.write.Writer._
import pru.write.{Action,JSON}

object Output {

  import Reportable._
  import JSON.literal

  def toJSON(k: Key[Any]): Action =
    JSON.Obj("label" -> literal(k.label), "id" -> literal(k.id.toString))

  def toJSON(ks: Seq[Key[Any]]): Action =
    JSON.list(ks.map(toJSON))

  def toJSON(d: Double): Action =
    if (d.isNaN ||
        d == Double.PositiveInfinity ||
        d == Double.NegativeInfinity)
      k("null")
    else k(d.toString)

  def toJSON(a: intelmedia.ws.monitoring.Stats): Action = JSON.Obj(
      "kind" -> literal("Stats"),
      "last" -> a.last.map(any).getOrElse(k("null")),
      "mean" -> toJSON(a.mean),
      "variance" -> toJSON(a.variance),
      "standardDeviation" -> toJSON(a.standardDeviation),
      "count" -> k(a.count.toString),
      "skewness" -> toJSON(a.skewness),
      "kurtosis" -> toJSON(a.kurtosis))

  def toJSON[A](r: Reportable[A]): Action = r match {
    case I(a) => k(a.toString)
    case B(a) => k(a.toString)
    case D(a) => k(a.toString)
    case S(s) => literal(s)
    case Stats(a) => toJSON(a)
    case _ => sys.error("unrecognized reportable: " + r)
  }

  def toJSON(m: Traversable[(Key[Any],Reportable[Any])]): Action =
    JSON.list { m.toList.map { case (key, v) =>
      JSON.Obj("label" -> literal(key.label),
          "id" -> literal(key.id.toString),
          "value" -> toJSON(v))
    }}

  /** Format a duration like `62 seconds` as `0hr 1m 02s` */
  def hoursMinutesSeconds(d: Duration): String = {
    val hours = d.toHours
    val minutes = ((d.toMinutes minutes)- (hours hours)).toMinutes
    val seconds = ((d.toSeconds seconds) - (hours hours) - (minutes minutes)).toSeconds
    s"${hours}hr ${minutes}m ${seconds}s"
  }

  /**
   * Write a server-side event stream (http://www.w3.org/TR/eventsource/)
   * of the given metrics to the `Writer`. This will block the calling
   * thread indefinitely.
   */
  def eventsToSSE(events: Process[Task, (Key[Any], Reportable[Any])],
                  sink: java.io.Writer): Unit =
    events.map(kv => s"event: ${toJSON(kv._1)}\ndata: ${toJSON(kv._2)}\n")
          .intersperse("\n")
          .map(writeTo(sink))<commit_msg>made the event name static so that JS handlers can be registered against it as per the origional intent of SSE<commit_after>package intelmedia.ws.monitoring

import java.util.concurrent.ExecutorService
import scala.concurrent.duration._
import scalaz.concurrent.{Strategy,Task}
import scalaz.stream._
import pru.write.Writer._
import pru.write.{Action,JSON}

object Output {

  import Reportable._
  import JSON.literal

  def toJSON(k: Key[Any]): Action =
    JSON.Obj("label" -> literal(k.label), "id" -> literal(k.id.toString))

  def toJSON(ks: Seq[Key[Any]]): Action =
    JSON.list(ks.map(toJSON))

  def toJSON(d: Double): Action =
    if (d.isNaN ||
        d == Double.PositiveInfinity ||
        d == Double.NegativeInfinity)
      k("null")
    else k(d.toString)

  def toJSON(a: intelmedia.ws.monitoring.Stats): Action = JSON.Obj(
      "kind"              -> literal("Stats"),
      "last"              -> a.last.map(any).getOrElse(k("null")),
      "mean"              -> toJSON(a.mean),
      "variance"          -> toJSON(a.variance),
      "standardDeviation" -> toJSON(a.standardDeviation),
      "count"             -> k(a.count.toString),
      "skewness"          -> toJSON(a.skewness),
      "kurtosis"          -> toJSON(a.kurtosis))

  def toJSON[A](r: Reportable[A]): Action = r match {
    case I(a) => k(a.toString)
    case B(a) => k(a.toString)
    case D(a) => k(a.toString)
    case S(s) => literal(s)
    case Stats(a) => toJSON(a)
    case _ => sys.error("unrecognized reportable: " + r)
  }

  def toJSON(m: Traversable[(Key[Any],Reportable[Any])]): Action =
    JSON.list { m.toList.map { case (key, v) =>
      JSON.Obj("label" -> literal(key.label),
          "id" -> literal(key.id.toString),
          "value" -> toJSON(v))
    }}

  def toJSON(tuple: (Key[Any], Reportable[Any])): Action = 
    JSON.Obj(
      "label"      -> literal(tuple._1.label), 
      "id"         -> literal(tuple._1.id.toString),
      "reportable" -> toJSON(tuple._2))

  /** Format a duration like `62 seconds` as `0hr 1m 02s` */
  def hoursMinutesSeconds(d: Duration): String = {
    val hours = d.toHours
    val minutes = ((d.toMinutes minutes)- (hours hours)).toMinutes
    val seconds = ((d.toSeconds seconds) - (hours hours) - (minutes minutes)).toSeconds
    s"${hours}hr ${minutes}m ${seconds}s"
  }

  /**
   * Write a server-side event stream (http://www.w3.org/TR/eventsource/)
   * of the given metrics to the `Writer`. This will block the calling
   * thread indefinitely.
   */
  def eventsToSSE(events: Process[Task, (Key[Any], Reportable[Any])],
                  sink: java.io.Writer): Unit = 
    events.map(kv => s"event: reportable\ndata: ${toJSON(kv)}\n")
          .intersperse("\n")
          .map(writeTo(sink))
{'fraction_non_alphanumeric': 0.16741463414634147, 'fraction_numerical': 0.0037073170731707315, 'mean_word_length': 2.8599397590361444, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, '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': '6794752', 'n_tokens_mistral': 1801, 'n_tokens_neox': 1735, 'n_words': 381}