Algorithm: Fixed and Sliding Windows in Kotlin

partition.md

• partitionRanges.kt: partition() implements a function for fixed and sliding windows suited for lists (not sequences) using indices/ranges.

• partitionRecursion.kt: partitionRec() and partitionTailrec() implement such functions suited for lists using recursion and tail recursion (very inefficient). The concatenation of a lot of sequences using .plus() (or .flatten()) results in a StackOverflowException.

• partitionIterator.kt: batch() and slide() implement such functions suited for lists and sequences (similar to the prototype implementation). Small "problem" here is that source.iterator() introduces mutual state. Does not use RingBuffer or LinkedList, instead replaces the whole List.

• partitionZipDrop.kt: A sliding window for pairs with an offset of one, can be implemented ad hoc with this simple variant which uses zip(). This however does not work with sequenceOf(...).constrainOnce():

partitionIterator.kt

// Fixed `batch()` and sliding `slide()` windows suited for lists and sequences.

fun <T> Iterable<T>.batch(size: Int): Sequence<List<T>> {
    return BatchingSequence(this, size)
}

fun <T> Sequence<T>.batch(size: Int): Sequence<List<T>> {
    return BatchingSequence(this.asIterable(), size)
}

fun <T> Iterable<T>.slide(size: Int,
                          step: Int = 1): Sequence<List<T>> {
    return SlidingSequence(this, size, step)
}

fun <T> Sequence<T>.slide(size: Int,
                          step: Int = 1): Sequence<List<T>> {
    return SlidingSequence(this.asIterable(), size, step)
}

internal class BatchingSequence<out T>(val source: Iterable<T>,
                                       val batchSize: Int) : Sequence<List<T>> {
    override fun iterator(): Iterator<List<T>> = object : AbstractIterator<List<T>>() {
        private val iterator = if (batchSize > 0) source.iterator() else emptyList<T>().iterator()

        override fun computeNext() = when {
            iterator.hasNext() -> {
                val next = iterator.asSequence().take(batchSize).toList()
                setNext(next)
            }
            else -> done()
        }
    }
}

internal class SlidingSequence<out T>(val source: Iterable<T>,
                                      val slideSize: Int,
                                      val slideStep: Int) : Sequence<List<T>> {
    override fun iterator(): Iterator<List<T>> = object : AbstractIterator<List<T>>() {
        private val iterator = if (slideSize > 0) source.iterator() else emptyList<T>().iterator()
        private var buffer = listOf<T>()

        override fun computeNext() = when {
            iterator.hasNext() -> {
                buffer = buffer.drop(slideStep).let {
                    it + iterator.asSequence().take(slideSize - it.size)
                }
                setNext(buffer)
                
                //val next = buffer.drop(slideStep) + iterator.asSequence()
                //    .take(slideSize - Math.max(buffer.size - slideStep, 0))
                //buffer = next
                //setNext(next)
            }
            else -> done()
        }
    }
}

partitionRanges.kt

// Fixed and sliding windows for lists (not sequences) using indices/ranges.

fun <T> List<T>.partition(size: Int,
                          step: Int = size): Sequence<List<T>> =
    partitionRanges(size, step)
        .takeWhile { it.last in indices }
        .map { slice(it) }

internal fun partitionRanges(size: Int,
                             step: Int = size): Sequence<IntRange> =
    generateSequence(0) { it + step }
        .map { it .. it + size - 1 }

partitionRecursion.kt

// Recursive and tail recursive implementation of fixed and sliding windows for lists. Very inefficient.

fun <T> partitionRec(source: Iterable<T>,
                     size: Int,
                     step: Int): Sequence<List<T>> {
    val part = source.take(size)
    val rest = source.drop(step)
    return when {
        part.size == size -> sequenceOf(part) + partitionRec(rest, size, step)
        else -> sequenceOf()
    }
}

tailrec fun <T> partitionTailrec(source: Iterable<T>,
                                 result: Sequence<List<T>> = sequenceOf(),
                                 size: Int,
                                 step: Int): Sequence<List<T>> {
    val part = source.take(size)
    val rest = source.drop(step)
    return when {
        part.size == size -> partitionTailrec(rest, result + sequenceOf(part), size, step)
        else -> result + sequenceOf()
    }
}

partitionZipDrop.kt

fun main(args: Array<String>) {
    val itemSeq = sequenceOf(1, 2, 3, 4, 5) //.constrainOnce()
    val itemPairs = itemSeq.zip(itemSeq.drop(1))
    val itemPairPairs = itemSeq.zip(itemSeq.drop(1)).zip(itemSeq.drop(2))

    println(itemSeq.toList()) // [1, 2, 3, 4, 5]
    println(itemPairs.toList()) // [(1, 2), (2, 3), (3, 4), (4, 5)]
    println(itemPairPairs.toList()) // [((1, 2), 3), ((2, 3), 4), ((3, 4), 5)]
}