Usar Scalaz Stream para analizar la tarea(reemplazando Scalaz Iteratees)

transducer-machines scalaz-stream

Question

Introducción

Utilizo las iteraciones de Scalaz 7 en una serie de proyectos, principalmente para procesar archivos de gran tamaño. Me gustaría comenzar a cambiar a las transmisiones de Scalaz, que están diseñadas para reemplazar el paquete iteratee (que francamente falta muchas piezas y es un poco doloroso de usar).

Las transmisiones se basan en machines (otra variación de la idea iteratee), que también se han implementado en Haskell. He usado un poco la biblioteca de máquinas Haskell, pero la relación entre las máquinas y las transmisiones no es completamente obvia (para mí, al menos), y la documentación para la biblioteca de secuencias aún es un poco escasa .

Esta pregunta se trata de una simple tarea de análisis que me gustaría ver implementada usando streams en lugar de iteratees. Contestaré la pregunta yo mismo si nadie más me gana, pero estoy seguro de que no soy el único que está haciendo (o al menos está considerando) esta transición, y dado que necesito trabajar en este ejercicio de todos modos, Pensé que podría hacerlo en público.

Tarea

Supongo que tengo un archivo que contiene oraciones que han sido tokenizadas y etiquetadas con partes del discurso:

no UH
, ,
it PRP
was VBD
n''t RB
monday NNP
. .

the DT
equity NN
market NN
was VBD
illiquid JJ
. .

Hay un token por línea, las palabras y las partes del discurso están separadas por un solo espacio, y las líneas en blanco representan los límites de las frases. Quiero analizar este archivo y devolver una lista de oraciones, que también podríamos representar como listas de tuplas de cadenas:

List((no,UH), (,,,), (it,PRP), (was,VBD), (n''t,RB), (monday,NNP), (.,.))
List((the,DT), (equity,NN), (market,NN), (was,VBD), (illiquid,JJ), (.,.)

Como de costumbre, queremos fallar con elegancia si pulsamos las entradas no válidas o las excepciones de lectura de archivos, no queremos tener que preocuparnos por cerrar los recursos manualmente, etc.

Una solución iteratee

Primero para algunas cosas generales de lectura de archivos (que realmente deberían ser parte del paquete iteratee, que actualmente no proporciona nada remotamente de este alto nivel):

import java.io.{ BufferedReader, File, FileReader }
import scalaz._, Scalaz._, effect.IO
import iteratee.{ Iteratee => I, _ }

type ErrorOr[A] = EitherT[IO, Throwable, A]

def tryIO[A, B](action: IO[B]) = I.iterateeT[A, ErrorOr, B](
  EitherT(action.catchLeft).map(I.sdone(_, I.emptyInput))
)

def enumBuffered(r: => BufferedReader) = new EnumeratorT[String, ErrorOr] {
  lazy val reader = r
  def apply[A] = (s: StepT[String, ErrorOr, A]) => s.mapCont(k =>
    tryIO(IO(Option(reader.readLine))).flatMap {
      case None       => s.pointI
      case Some(line) => k(I.elInput(line)) >>== apply[A]
    }
  )
}

def enumFile(f: File) = new EnumeratorT[String, ErrorOr] {
  def apply[A] = (s: StepT[String, ErrorOr, A]) => tryIO(
    IO(new BufferedReader(new FileReader(f)))
  ).flatMap(reader => I.iterateeT[String, ErrorOr, A](
    EitherT(
      enumBuffered(reader).apply(s).value.run.ensuring(IO(reader.close()))
    )
  ))
}

Y luego nuestro lector de oraciones:

def sentence: IterateeT[String, ErrorOr, List[(String, String)]] = {
  import I._

  def loop(acc: List[(String, String)])(s: Input[String]):
    IterateeT[String, ErrorOr, List[(String, String)]] = s(
    el = _.trim.split(" ") match {
      case Array(form, pos) => cont(loop(acc :+ (form, pos)))
      case Array("")        => cont(done(acc, _))
      case pieces           =>
        val throwable: Throwable = new Exception(
          "Invalid line: %s!".format(pieces.mkString(" "))
        )

        val error: ErrorOr[List[(String, String)]] = EitherT.left(
          throwable.point[IO]
        )

        IterateeT.IterateeTMonadTrans[String].liftM(error)
    },
    empty = cont(loop(acc)),
    eof = done(acc, eofInput)
  )
  cont(loop(Nil))
}

Y finalmente nuestra acción de análisis:

val action =
  I.consume[List[(String, String)], ErrorOr, List] %=
  sentence.sequenceI &=
  enumFile(new File("example.txt"))

Podemos demostrar que funciona:

scala> action.run.run.unsafePerformIO().foreach(_.foreach(println))
List((no,UH), (,,,), (it,PRP), (was,VBD), (n''t,RB), (monday,NNP), (.,.))
List((the,DT), (equity,NN), (market,NN), (was,VBD), (illiquid,JJ), (.,.))

Y terminamos.

Lo que quiero

Más o menos el mismo programa implementado usando transmisiones de Scalaz en lugar de iterar.

Answer 1

Una solución scalaz-stream:

import scalaz.std.vector._
import scalaz.syntax.traverse._
import scalaz.std.string._

val action = linesR("example.txt").map(_.trim).
  splitOn("").flatMap(_.traverseU { s => s.split(" ") match {
    case Array(form, pos) => emit(form -> pos)
    case _ => fail(new Exception(s"Invalid input $s"))
  }})

Podemos demostrar que funciona:

scala> action.collect.attempt.run.foreach(_.foreach(println))
Vector((no,UH), (,,,), (it,PRP), (was,VBD), (n''t,RB), (monday,NNP), (.,.))
Vector((the,DT), (equity,NN), (market,NN), (was,VBD), (illiquid,JJ), (.,.))

Y terminamos.

La función traverseU es un combinador Scalaz común. En este caso, se está utilizando para atravesar, en la mónada Process , la oración Vector generada por splitOn . Es equivalente a un map seguido de una sequence .