Como encontrar áreas retangulares de mesmo valor de um determinado tamanho em uma matriz com mais eficiência?

Meu problema é muito simples, mas ainda não encontrei uma implementação eficiente.

Suponha que exista uma matriz A como esta:

0 0 0 0 0 0 0
4 4 2 2 2 0 0
4 4 2 2 2 0 0
0 0 2 2 2 1 1
0 0 0 0 0 1 1

Agora, quero encontrar todas as posições iniciais de áreas retangulares nessa matriz que tenham um determinado tamanho. Uma área é um subconjunto de A onde todos os números são iguais.

Digamos largura = 2 e altura = 3. Existem 3 áreas com esse tamanho:

2 2   2 2   0 0
2 2   2 2   0 0
2 2   2 2   0 0

O resultado da chamada de função seria uma lista de posições iniciais (x, y começando com 0) dessas áreas.

List((2,1),(3,1),(5,0))

A seguir está minha implementação atual. "Áreas" são chamadas "superfícies" aqui.

case class Dimension2D(width: Int, height: Int)
case class Position2D(x: Int, y: Int)

def findFlatSurfaces(matrix: Array[Array[Int]], surfaceSize: Dimension2D): List[Position2D] = {

    val matrixWidth = matrix.length
    val matrixHeight = matrix(0).length
    var resultPositions: List[Position2D] = Nil

    for (y <- 0 to matrixHeight - surfaceSize.height) {
        var x = 0
        while (x <= matrixWidth - surfaceSize.width) {
            val topLeft = matrix(x)(y)
            val topRight = matrix(x + surfaceSize.width - 1)(y)
            val bottomLeft = matrix(x)(y + surfaceSize.height - 1)
            val bottomRight = matrix(x + surfaceSize.width - 1)(y + surfaceSize.height - 1)
            // investigate further if corners are equal
            if (topLeft == bottomLeft && topLeft == topRight && topLeft == bottomRight) {
                breakable {
                    for (sx <- x until x + surfaceSize.width;
                         sy <- y until y + surfaceSize.height) {
                        if (matrix(sx)(sy) != topLeft) {
                            x = if (x == sx) sx + 1 else sx 
                            break
                        }
                    }
                    // found one!       
                    resultPositions ::= Position2D(x, y)
                    x += 1
                }
            } else if (topRight != bottomRight) {
                // can skip x a bit as there won't be a valid match in current row in this area
                x += surfaceSize.width 
            } else {
                x += 1
            }
        }   
    }
    return resultPositions
}

Eu já tentei incluir algumas otimizações, mas tenho certeza de que existem soluções muito melhores. Existe uma função matlab para ela que eu possa portar? Também estou me perguntando se esse problema tem seu próprio nome, pois eu não sabia exatamente o que procurar no Google.

Obrigado por pensar nisso! Estou animado para ver suas propostas ou soluções :)

EDITAR: As dimensões da matriz em meu aplicativo variam de 300 x 300 a 3000 x 3000, aproximadamente. Além disso, o algoritmo será chamado apenasuma vez para a mesma matriz. O motivo é que a matriz sempre será alterada posteriormente (aproximadamente 1 a 20%).

RESULTADOS

Eu implementei os algoritmos de Kevin, Nikita e Daniel e os avaliei no meu ambiente de aplicativos, ou seja, nenhum benchmark sintético isolado aqui, mas tomei um cuidado especial para integrar todos os algoritmos da maneira mais eficiente possível, o que foi especialmente importante para a abordagem de Kevin, pois usa genéricos (ver abaixo).

Primeiro, os resultados brutos, usando Scala 2.8 e jdk 1.6.0_23. Os algoritmos foram executados várias centenas de vezes como parte da solução de um problema específico do aplicativo. "Duração" indica o tempo total necessário até o algoritmo do aplicativo terminar (é claro, sem a inicialização da jvm etc.). Minha máquina é um Core 2 Duo de 2,8 GHz com 2 núcleos e 2 gig de memória, -Xmx800M foram dados à JVM.

NOTA IMPORTANTE: Acho que minha configuração de benchmark não é realmente justa para algoritmos paralelos como o de Daniel. Isso ocorre porque o aplicativo já está calculando multithread. Portanto, os resultados aqui provavelmente mostram apenas um equivalente à velocidade de rosca única.

Tamanho da matriz 233x587:

                  duration | JVM memory | avg CPU utilization
original O(n^4) | 3000s      30M          100%  
original/-server| 840s       270M         100%
Nikita O(n^2)   | 5-6s       34M          70-80%
Nikita/-server  | 1-2s       300M         100%
Kevin/-server   | 7400s      800M         96-98%
Kevin/-server** | 4900s      800M         96-99%
Daniel/-server  | 240s       360M         96-99%

** com @specialized, para fazergenéricos mais rápido evitando o apagamento do tipo

Tamanho da matriz 2000x3000:

                  duration | JVM memory | avg CPU utilization
original O(n^4) | too long   100M         100%  
Nikita O(n^2)   | 150s       760M         70%
Nikita/-server  | 295s (!)   780M         100%
Kevin/-server   | too long, didn't try

Primeiro, uma pequena nota na memória. A opção -server JVM usa consideravelmente mais memória com a vantagem de mais otimizações e, em geral, uma execução mais rápida. Como você pode ver na 2ª tabela, o algoritmo da Nikita é mais lento com a opção -server, que obviamente se deve ao limite de memória. Suponho que isso também diminua o algoritmo de Kevin, mesmo para a matriz pequena, pois a abordagem funcional está usando muito mais memória. Para eliminar o fator de memória, eu também tentei uma vez com uma matriz 50x50 e, em seguida, Kevin levou 5 segundos e 0 segundo de Nikita (bem, quase 0). Portanto, em qualquer caso, ainda é mais lento e não apenas por causa da memória.

Como você pode ver pelos números, obviamente usarei o algoritmo da Nikita porque é muito rápido e isso é absolutamente necessário no meu caso. Também pode ser paralelo facilmente, como Daniel apontou. A única desvantagem é que não é realmente o caminho da escória.

No momento, o algoritmo de Kevin provavelmente é um pouco complexo demais e, portanto, lento, mas tenho certeza de que há mais otimizações possíveis (veja os últimos comentários em sua resposta).

Com o objetivo de transformar diretamente o algoritmo de Nikita em estilo funcional, Daniel encontrou uma solução que já é bastante rápida e, como ele diz, seria ainda mais rápida se pudesse usar o scanRight (veja os últimos comentários em sua resposta).

Qual é o próximo?

No lado tecnológico: aguardar o Scala 2.9, ScalaCL e fazer benchmarks sintéticos para obter velocidades brutas.

Meu objetivo em tudo isso é ter código funcional, mas apenas se não estiver sacrificando muita velocidade.

Escolha da resposta:

Quanto à escolha de uma resposta, gostaria de marcar os algoritmos de Nikita e Daniel como respostas, mas tenho que escolher uma. O título da minha pergunta incluía "da maneira mais eficiente", e uma é a mais rápida em imperativa e a outra em estilo funcional. Embora essa pergunta tenha a tag Scala, escolhi o algoritmo imperativo da Nikita como 2s vs. 240s ainda é muita diferença para eu aceitar. Tenho certeza de que a diferença ainda pode ser um pouco reduzida, alguma idéia?

Então, muito obrigado a todos! Embora eu não use os algoritmos funcionaisainda, Obtive muitas novas idéias sobre o Scala e acho que lentamente entendo toda a loucura funcional e seu potencial. (é claro, mesmo sem fazer muita programação funcional, o Scala é muito mais agradável que o Java ... esse é outro motivo para aprendê-lo)

questionAnswers(3)

yourAnswerToTheQuestion