Optimierungsproblem für Datenkompression

[darktrym]

Ich hatte mich vor einigen Monden mal mit der Tile-Kompression alter Spielkonsolen, eigentlich nur SMS, beschäftigt. Einfache(schnelle) und eff. Verfahren bauen auf RLE auf. Hängengeblieben bin ich da bei der eff. Umsetzung von der "Phantasy Star"-RLE..

Das Kodierschema sieht wie folgt aus:

%0nnnnnnn dd ; run of n consecutive identical bytes, value dd
%1nnnnnnn dd dd dd... ; run of n consecutive non-identical bytes; values follow
%00000000 ; end of bitplane

Auf den ersten Blick trivial zu implementieren, aber will man die beste Kompressionsrate erreichen, wirds kompliziert. Offensichtlich ist es entscheidend wann der Wechsel zwischen den beiden Modi erfolgt oder besser gesagt der sollte nicht allzu oft sein, da ungleiche Zeichenfolgen stets ein zusätzliche Byte als Präfix benötigen.
Also wie finde ich da das Optimum?

[Sirius3]

@darktrym: komprimiere einfach A B ... B C mit unterschiedlich vielen Bs:

Code: Alles auswählen

81 0A 0? 0B 81 0C

Du siehst, Du brauchst 3 Bytes für beliebig viele Bs.

Code: Alles auswählen

8? 0A 0B .. 0B 0C

Ohne Kompression brauchst Du n Bytes für n Bs.
Solange also n <= 3 ist, lohnt sich eine Kompression nicht. Außer natürlich, der Block an unkomprimierten Bytes ist größer als 127. Wenn durch die Aufspaltung die Anzahl der benötigten Blöcke nicht steigt, ist es günstiger bei n=3 zu komprimieren.

[darktrym]

Also bei n=3 lohnt sich das schon, Beispiel:
A A A B B B = 0bin(3) A 0bin(3) B .

Das Problem sind lange ungleiche Sequenzen unterbrochen durch kurze gleiche.
X Y X Y .. X X X Y X Y X. Richtig ist da ab n=4 für den Modiwechsel, die Kodierung ist besser denn es gilt:

1bin(ANZAHL) Seq ist besser als 1bin(ANZAHL_SEQ1) SEQ1 0bin(3) X 1bin(ANZAHL_SEQ2) SEQ2
Sequenzen länger 127 müssen dann aufgeteilt werden, was die Sache nicht gerade einfacher macht, das Optimum zu finden.
Dazu müsste man weit in die Zukunft schauen oder irgendwas aus dem Bereich der dynamischen Programmierung finden, was passt.

[BlackJack]

@darktrym: Ich sehe da jetzt aber immer noch nicht was man da grossartig herumprobieren muss. Einfach immer bei drei oder mehr gleichen Bytes einen „run“ kodieren und gut ist. Kannst Du ein konkretes Beispiel geben bei dem man anders eine kürzere Variante erhalten würde?

[bords0]

Ein X, gefolgt von 128 Y.

[BlackJack]

@bords0: Okay, die Variante ohne herumprobieren wäre:

Code: Alles auswählen

81 XX 7f YY 81 YY

Was wäre da jetzt kürzer?

[bords0]

Sorry, dachte das wäre offensichtlich:

Code: Alles auswählen

82 XX YY 7f YY

[BlackJack]

@bords0: Jetzt ja. Ich schiebe das mal auf die Uhrzeit meines letzten Beitrags und das der letzte Kaffee sehr lange her war.

Dieser Fall liesse sich noch ”greedy” lösen in dem man nicht-runs nicht sofort raus schreibt, sondern wartet bis man den nächsten Run zusammen hat und eventuell noch ein oder zwei Bytes davon mit in den nicht-run davor stecken muss.

[bords0]

@darktrym: Das Folgende sollte die optimale Lösung finden. Benutzt dynamische Programmierung. Rechnet von hinten, und kennt daher für alle Suffixe bereits die optimale Lösung.

Code: Alles auswählen

def RLE(A):
    optimal = [None] * len(A)
    optimal.append((0, "dummy", 0, [])) # total length of encoded sequence, runtype (0 or 128), number of values, list of values

    for start in range(len(A))[::-1]:
        candidates = []
        # non-identical runs
        for run_length in range(min(len(A) - start, 127)): 
            candidates.append((2 + run_length + optimal[start + run_length + 1][0], 128, run_length + 1, A[start:start+run_length+1]))
        # identical runs
        for run_length in range(1, min(len(A) - start, 127)): 
            if A[start] != A[start + run_length]:
                break
            candidates.append((2 + optimal[start + run_length + 1][0], 0, run_length + 1, A[start:start+1]))
        optimal[start] = min(candidates)
        
    steps = []
    pos = 0
    while pos < len(A):
        length, run_type, run_length, values = optimal[pos]
        steps.append(run_type + run_length)
        steps.extend(values)
        pos += run_length
    return steps

Beispiele:

Code: Alles auswählen

>>> RLE([100] + 128*[101])
[130, 100, 101, 127, 101]
>>> RLE([100] + 128*[101] + [100])
[129, 100, 127, 101, 130, 101, 100]
>>> RLE([100] + 128*[101] + [100,100])
[130, 100, 101, 127, 101, 2, 100]

[BlackJack]

Ich habe mich mal an einem Algorithmus versucht der die Eingabedaten sequentiell liest und schreibt, und den man so implementieren könnte das er mit konstantem zusätzlichen Speicher auskommt:

Code: Alles auswählen

from itertools import groupby


def iter_runs(data):
    for value, group in groupby(data):
        yield value, sum(1 for _ in group)


def rle(data):
    result = list()
    # 
    # This list collects bytes that will be encoded as non-runs i.e. prefixed
    # by a count byte with a set bit 7.  Although it can become up to
    # `len(data)` elements in length, this algorithm can be adapted to use
    # at most 127 entries.
    # 
    # It may even be possible to use the result list as buffer for these
    # entries by storing a dummy count byte which will be fixed up when the
    # actual count is determined.
    # 
    non_run = list()

    def copy_non_run():
        """Copy the content of the `non_run` buffer to the result."""
        for i in xrange(0, len(non_run), 127):
            chunk = non_run[i:i + 127]
            result.append(128 | len(chunk))
            result.extend(chunk)

    for value, count in iter_runs(data):
        if count == 1:
            non_run.append(value)
        else:
            # 
            # Check if there is one single value left which cannot be encoded
            # in one (count byte, value) pair.
            # 
            # If the single value does fit into a non-run then append it
            # before the encoded run, otherwise put it into the non-run buffer
            # after encoding the run.
            # 
            one_byte_left = count % 127 == 1
            if one_byte_left:
                if len(non_run) % 127 != 0:
                    non_run.append(value)
                    one_byte_left = False
                count -= 1
            # 
            # Flush the collected non-run bytes so far.
            # 
            copy_non_run()
            non_run = [value] if one_byte_left else []
            # 
            # Encode the run.
            # 
            while count:
                run_count = min(127, count)
                result.extend([run_count, value])
                count -= run_count
    # 
    # Flush the possibly remaining non-run bytes.
    # 
    copy_non_run()
    return result

[bords0]

@Blackjack: Das ist ein NameError (iter_runs).

[BlackJack]

@bords0: Ups, wie peinlich. Habe die Funktion reineditiert.

[bords0]

@BlackJack: Jetzt läufts. Aber optimale Ergebnisse liefert es nicht immer. Im Vergleich mit meiner RLE-Funktion oben:

Code: Alles auswählen

In [2]: rle([0, 1, 1, 0])
Out[2]: [129, 0, 2, 1, 129, 0]

In [3]: RLE([0, 1, 1, 0])
Out[3]: [132, 0, 1, 1, 0]

Ich glaube, dass es ganz schön komplex ist, das Optimum durch "Vorausschauen" herauszufinden.