<commit_before>/*
 Copyright 2009 <NAME>, <NAME>

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
*/
package scalanlp.stage;

import scala.reflect.Manifest;

import scalanlp.collection.immutable.HTMap;

/**
 * Represents a batch of elements that can be treated as just
 * an iterable of V.  Use the static constructors in Batch
 * to create instances.
 * 
 * @author dramage
 */
trait Batch[+V] {
  /** An iterable of items ordered by their number. */
  def items : Iterable[Item[V]];
  
  /** An iterable of the values in this map (unboxes items). */
  def values : Iterable[V] =
    items.projection.map(_.value);

  /**
   * Returns an iterable of options of values in this map of
   * length this.size.  The iterable returns Some(item.value)
   * for items present in this.items, but returns None for
   * offsets that have no corresponding item.
   */
  def options : Iterable[Option[V]] = new Iterable[Option[V]] {
    override def elements = new Iterator[Option[V]] {
      var itemNum = 0;
      val itemIter = items.elements.buffered.advanced;
      
      override def hasNext =
        itemNum < size;
      
      override def next : Option[V] = {
        val rv = if (itemIter.hasNext && itemIter.peek(0).number == itemNum)
          Some(itemIter.next.value) else None;
        itemNum += 1;
        return rv;
      }
    }
  }
    
  
  /**
   * The total number of items undelrying this batch, which is 
   * greater* than or equal to the size of this.items and
   * this.values (because some items may have been filtered
   * from another underlying batch).
   */
  def size : Int;
  
  /** Transforms the items of the batch according to the given function. */
  def map[O](f : V => O) : Batch[O] = {
    def mapper(item : Item[V]) = {
      try {
        item.map(f)
      } catch {
        case ex : Throwable =>
          throw new BatchException(item, ex);
      }
    }
    Batch.fromItems[O](items.projection.map(mapper), size);
  }
  
  /** Filters out items from the batch according to the given funciton. */
  def filter(f : V => Boolean) : Batch[V] = {
    def filterer(item : Item[V]) = {
      try {
        f(item.value);
      } catch {
        case ex : Throwable =>
          throw new BatchException(item, ex);
      }
    }
    Batch.fromItems(items.projection.filter(filterer), size);
  }
  
  /** Takes the first n elements. */
  def take(n : Int) : Batch[V] =
    Batch.fromItems(items.projection.takeWhile(_.number < n), n);

  /** Drops the first n elements. */
  def drop(n : Int) : Batch[V] =
    Batch.fromItems(items.projection.dropWhile(_.number < n), size-n);
  
  /** Zips together two batches. */
  def zip[O](that : Batch[O]) = {
    Batch.Zip(this,that).map(
      seq => (seq(0).asInstanceOf[Option[V]],
              seq(1).asInstanceOf[Option[O]]
    ));
  }
}

/**
 * An exception thrown by Batch when a map or filter filter fails on
 * a given row.
 * 
 * @author dramage
 */
class BatchException(item : Item[_], cause : Throwable)
extends RuntimeException("Unable to process " + item +
                           " item " + item.number, cause);

/**
 * Static constructors for creating batches.
 */
object Batch {
  def fromItems[V](inItems : Iterable[Item[V]], numItems : Int) = new Batch[V] {
    override def size = numItems;
    override def items = inItems;
  }
  
  def fromIterable[V](inItems : Iterable[V]) = new Batch[V] {
  /** Number of items in this batch. */
  // TODO: in scala 2.8.0 this can just delegate to items.size
  override lazy val size = {
    var s = 0;
    for (item <- items.elements) {
      s += 1;
    }
    s;
  }

    override def items = new Iterable[Item[V]] {
      override def elements = {
        for ((v,i) <- inItems.elements.zipWithIndex) yield
          Item(i, v);
      }
    }
  }
  
  /** Zips together two batches. */
  def zip[A,B](batchA : Batch[A], batchB : Batch[B]) =
    (batchA zip batchB);
  
  /** Zips together three batches. */
  def zip[A,B,C](batchA : Batch[A], batchB : Batch[B], batchC : Batch[C])
  : Batch[(Option[A],Option[B],Option[C])] = {
    Batch.Zip(batchA,batchB,batchC).map(
      seq => (seq(0).asInstanceOf[Option[A]],
              seq(1).asInstanceOf[Option[B]],
              seq(2).asInstanceOf[Option[C]]
    ));
  }
  
  /**
   * A single batch consisting of items selected from the incoming list of batches
   * by matching on item number.  This batch iterates over Array[Option[V]] where
   * the sequence order matches the order of the batches passed as input, and<commit_msg>added .strict method on Batch<commit_after>/*
 Copyright 2009 <NAME>, <NAME>

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
*/
package scalanlp.stage;

import scala.reflect.Manifest;

import scalanlp.collection.immutable.HTMap;

/**
 * Represents a batch of elements that can be treated as just
 * an iterable of V.  Use the static constructors in Batch
 * to create instances.
 * 
 * @author dramage
 */
trait Batch[+V] {
  /** An iterable of items ordered by their number. */
  def items : Iterable[Item[V]];
  
  /** An iterable of the values in this map (unboxes items). */
  def values : Iterable[V] =
    items.projection.map(_.value);

  /**
   * Returns an iterable of options of values in this map of
   * length this.size.  The iterable returns Some(item.value)
   * for items present in this.items, but returns None for
   * offsets that have no corresponding item.
   */
  def options : Iterable[Option[V]] = new Iterable[Option[V]] {
    override def elements = new Iterator[Option[V]] {
      var itemNum = 0;
      val itemIter = items.elements.buffered.advanced;
      
      override def hasNext =
        itemNum < size;
      
      override def next : Option[V] = {
        val rv = if (itemIter.hasNext && itemIter.peek(0).number == itemNum)
          Some(itemIter.next.value) else None;
        itemNum += 1;
        return rv;
      }
    }
  }
    
  
  /**
   * The total number of items undelrying this batch, which is 
   * greater* than or equal to the size of this.items and
   * this.values (because some items may have been filtered
   * from another underlying batch).
   */
  def size : Int;
  
  /**
   * Transforms the items of the batch according to the given function.
   * This is non-strict by default, i.e. the computation is done again
   * each time the returned batch is iterated.
   */
  def map[O](f : V => O) : Batch[O] = {
    def mapper(item : Item[V]) = {
      try {
        item.map(f)
      } catch {
        case ex : Throwable =>
          throw new BatchException(item, ex);
      }
    }
    Batch.fromItems[O](items.projection.map(mapper), size);
  }
  
  /** Filters out items from the batch according to the given funciton. */
  def filter(f : V => Boolean) : Batch[V] = {
    def filterer(item : Item[V]) = {
      try {
        f(item.value);
      } catch {
        case ex : Throwable =>
          throw new BatchException(item, ex);
      }
    }
    Batch.fromItems(items.projection.filter(filterer), size);
  }
  
  /** Takes the first n elements. */
  def take(n : Int) : Batch[V] =
    Batch.fromItems(items.projection.takeWhile(_.number < n), n);

  /** Drops the first n elements. */
  def drop(n : Int) : Batch[V] =
    Batch.fromItems(items.projection.dropWhile(_.number < n), size-n);
  
  /** Zips together two batches. */
  def zip[O](that : Batch[O]) = {
    Batch.Zip(this,that).map(
      seq => (seq(0).asInstanceOf[Option[V]],
              seq(1).asInstanceOf[Option[O]]
    ));
  }
  
  /** Creates a list-backed view of this batch (i.e. makes it strict). */
  def strict : Batch[V] =
    Batch.fromItems(items.toList, size);
}

/**
 * An exception thrown by Batch when a map or filter filter fails on
 * a given row.
 * 
 * @author dramage
 */
class BatchException(item : Item[_], cause : Throwable)
extends RuntimeException("Unable to process " + item +
                           " item " + item.number, cause);

/**
 * Static constructors for creating batches.
 */
object Batch {
  def fromItems[V](inItems : Iterable[Item[V]], numItems : Int) = new Batch[V] {
    override def size = numItems;
    override def items = inItems;
  }
  
  def fromIterable[V](inItems : Iterable[V]) = new Batch[V] {
    // TODO: in scala 2.8.0 this can just delegate to items.size
    private lazy val cachedSize = {
      var s = 0;
      for (item <- items.elements) {
        s += 1;
      }
      s;
    }
    
    override def size =
      cachedSize;

    override def items = new Iterable[Item[V]] {
      override def elements = {
        for ((v,i) <- inItems.elements.zipWithIndex) yield
          Item(i, v);
      }
    }
  }
  
  /** Zips together two batches. */
  def zip[A,B](batchA : Batch[A], batchB : Batch[B]) =
    (batchA zip batchB);
  
  /** Zips together three batches. */
  def zip[A,B,C](batchA : Batch[A], batchB : Batch[B], batchC : Batch[C])
  : Batch[(Option[A],Option[B],Option[C])] = {
    Batch.Zip(batchA,batchB,batchC).map(
      seq => (seq(0).asInstanceOf[Option[A]],
              seq(1).asInstanceOf[Option[B]],
              seq(2).asInstanceOf[Option[C]]
    ));
  }
  
  /**
   * A single batch consisting of items selected from the incoming list of batches
   * by matching on item number.  This batch iterates over Array[Option[V]] where
   * the sequence order matches the order of the batches passed as input, and
{'fraction_non_alphanumeric': 0.12027158098933075, 'fraction_numerical': 0.004073714839961203, 'mean_word_length': 2.6014669926650367, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, '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': '4727730', 'n_tokens_mistral': 3286, 'n_tokens_neox': 3211, 'n_words': 1057}