将 Scala 范围拆分为大小均匀的连续子范围

Question

如果我有一个 Range，如何将其拆分为一系列连续的子范围，其中指定了子范围（存储桶）的数量？如果没有足够的项目，应该省略空桶。

例如：

splitRange(1 to 6, 3) == Seq(Range(1,2), Range(3,4), Range(5,6))
splitRange(1 to 2, 3) == Seq(Range(1), Range(2))

一些额外的限制，排除了我见过的一些解决方案：

1. 大致均匀的存储桶大小 - 存储桶大小最多应变化 1
2. 输入范围的长度有时可能非常大，因此不应将范围具体化为序列（例如，不能使用grouped）
3. 这也意味着我们不会以循环方式将数字分配给存储桶，因为这样每个存储桶中的数字将不连续，因此不会形成范围
4. 理想情况下，子范围应按顺序生成，即 (1,2)(3,4)，而不是 (3,4)(1,2)

同事找到了解决办法here：

def splitRange(r: Range, chunks: Int): Seq[Range] = {
  if (r.step != 1) 
      throw new IllegalArgumentException("Range must have step size equal to 1")

  val nchunks = scala.math.max(chunks, 1)
  val chunkSize = scala.math.max(r.length / nchunks, 1)
  val starts = r.by(chunkSize).take(nchunks)
  val ends = starts.map(_ - 1).drop(1) :+ r.end
  starts.zip(ends).map(x => x._1 to x._2)
}

但是当 N 很小时，这会产生非常不均匀的桶大小，例如：

splitRange(1 to 14, 5)                          
//> Vector(Range(1, 2), Range(3, 4), Range(5, 6),
//|        Range(7, 8), Range(9, 10, 11, 12, 13, 14))
                              ^^^^^^^^^^^^^^^^^^^^^

Answer 1

浮点方法

一种方法是为每个存储桶生成一个小数（浮点）偏移量，然后通过压缩将它们转换为整数范围。空范围也需要使用collect 过滤掉。

def splitRange(r: Range, chunks: Int): Seq[Range] = {
  require(r.step == 1, "Range must have step size equal to 1")
  require(chunks >= 1, "Must ask for at least 1 chunk")

  val m = r.length.toDouble
  val chunkSize = m / chunks
  val bins = (0 to chunks).map { x => math.round((x.toDouble * m) / chunks).toInt }
  val pairs = bins zip (bins.tail)
  pairs.collect { case (a, b) if b > a => a to b }
}

（此解决方案的第一个版本存在舍入问题，因此无法处理 Int.MaxValue - 现在已根据下面 Rex Kerr 的递归浮点解决方案进行了修复）

另一种浮点方法是向下递归范围，每次都将头部移出范围，这样我们就不会错过任何元素。这个版本可以正确处理Int.MaxValue。

def splitRange(r: Range, chunks: Int): Seq[Range] = {
  require(r.step == 1, "Range must have step size equal to 1")
  require(chunks >= 1, "Must ask for at least 1 chunk")

  val chunkSize = r.length.toDouble / chunks

  def go(i: Int, r: Range, delta: Double, acc: List[Range]): List[Range] = {  
    if (i == chunks) r :: acc 
      // ensures the last chunk has all remaining values, even if error accumulates
    else {
      val s = delta + chunkSize
      val (chunk, rest) = r.splitAt(s.toInt)
      go(i + 1, rest, s - s.toInt, if (chunk.length > 0) chunk :: acc else acc)
    }
  }

  go(1, r, 0.0D, Nil).reverse
} 

也可以递归生成 (start,end) 对，而不是压缩它们。本文改编自 Rex Kerr 的answer to a similar question

def splitRange(r: Range, chunks: Int): Seq[Range] = {
  require(r.step == 1, "Range must have step size equal to 1")
  require(chunks >= 1, "Must ask for at least 1 chunk")

  val m = r.length
  val bins = (0 to chunks).map { x => math.round((x.toDouble * m) / chunks).toInt }
  def snip(r: Range, ns: Seq[Int], got: Vector[Range]): Vector[Range] = {
    if (ns.length < 2) got
    else {
      val (i, j) = (ns.head, ns.tail.head)
      snip(r.drop(j - i), ns.tail, got :+ r.take(j - i))
    }
  }
 snip(r, bins, Vector.empty).filter(_.length > 0)
}

整数方法

最后，我意识到这可以通过调整Bresenham's line-drawing algorithm 用纯整数运算来完成，这解决了一个基本等效的问题 - 如何仅使用整数运算在 y 行中均匀分配 x 像素！

我最初使用var 和ArrayBuffer 将伪代码转换为命令式解决方案，然后将其转换为尾递归解决方案：

def splitRange(r: Range, chunks: Int): List[Range] = {
  require(r.step == 1, "Range must have step size equal to 1")
  require(chunks >= 1, "Must ask for at least 1 chunk")

  val dy = r.length
  val dx = chunks

  @tailrec
  def go(y0:Int, y:Int, d:Int, ch:Int, acc: List[Range]):List[Range] = {
    if (ch == 0) acc
    else {
      if (d > 0) go(y0, y-1, d-dx, ch, acc)
      else go(y-1, y, d+dy, ch-1, if (y > y0) acc 
                                  else (y to y0) :: acc)
    }
  }

  go(r.end, r.end, dy - dx, chunks, Nil)
}

请参阅 Wikipedia 链接以获取完整说明，但本质上，该算法会在直线的斜率上曲折向上，或者添加 y 范围 dy 并减去 x 范围 dx。如果这些不完全划分，那么一个错误会累积，直到它完全划分，导致一些子范围中的额外像素。

splitRange(3 to 15, 5)                         
//> List(Range(3, 4), Range(5, 6, 7), Range(8, 9), 
//|      Range(10, 11, 12), Range(13, 14, 15))