x.scala

package brc

import monix.eval.Task
import monix.execution
import monix.reactive.Observable

import java.io.{BufferedReader, FileInputStream, InputStreamReader}
import java.text.DecimalFormat

object UsingMonix:
  val path = "/.../1brc/measurements.txt"

  case class Reading(station: String, temp: Double)
  case class Station(min: Double, max: Double, total: Double, num: Int):
    override def toString: String =
      val mean = total / num
      s"$min/${meanFormat.format(mean)}/$max"

  val meanFormat = DecimalFormat("#.#")
  def readReading(line: String) =
    val Array(station, temp) = line.split(';')
    Reading(station, temp.toDouble)

  def updateStation(data: Map[String, Station], reading: Reading) =
    data.updatedWith(reading.station): maybeData =>
      Some:
        maybeData.fold(Station(reading.temp, reading.temp, reading.temp, 1)): data =>
          val newMin = Math.min(reading.temp, data.min)
          val newMax = Math.max(reading.temp, data.max)
          val newTotal = data.total + reading.temp
          val newNum = data.num + 1
          data.copy(newMin, newMax, newTotal, newNum)

  def main(args: Array[String]) =
    def reader = new BufferedReader(new InputStreamReader(new FileInputStream(path), "UTF-8"))

    val summary = Observable
      .fromLinesReader(Task.eval(reader))
      .map(readReading)
      .foldLeftL(Map.empty)(updateStation)

    val result = summary.map: data =>
      data.toList
        .sortBy(_._1)
        .map: data =>
          s"${data._1}=${data._2}"
        .mkString(", ")

    import monix.execution.Scheduler.Implicits.global
    val s = result.runSyncUnsafe()
    println(s"{$s}")