ускорить гамма-сжатие Элиаса

Я написал несколько функций для сжатия и распаковки списка целых чисел, используя Гамма-кодирование Элиаса:

def _reverseBits(number: int) -> int:
    rev = 0
    while number:
        rev = rev << 1
        if number & 1:
            rev = rev ^ 1 # bitwise ^ 1 is faster than arithmetic + 1
        number = number >> 1
    return rev


def compress(numbers: List[int]) -> int:
    r'''Calculate the elias gamma encoded bitstream representation of the input list of numbers.

    >>> compress([5]) == 0b10100
    True
    >>> compress([15]) == 0b1111000
    True
    >>> compress([5, 15]) == 0b10100010100
    True
    '''
    bitstream = 0
    bitstream_length = 0
    previous_number = 0
    for number in numbers:
        delta = number - previous_number # store deltas, much smaller numbers
        N = delta.bit_length() - 1
        encoded_number = _reverseBits(delta) << N # reversed bitstream won't have leading zeroes problem in int container
        bitstream += encoded_number << bitstream_length
        bitstream_length += (N * 2 + 1)
        previous_number = number
    return bitstream


def decompress(bitstream: int) -> List[int]:
    r'''Convert the elias-encoded bitstream into a list of numbers.

    >>> decompress(0b10100010100)
    [5, 15]
    '''
    numbers = []
    cumulative_sum = 0
    while bitstream:
        N = 1
        while bitstream & 1 == 0: # count Elias leading zeroes
            bitstream = bitstream >> 1
            N += 1
        mask = (1 << N) - 1
        number = _reverseBits(bitstream & mask)
        number = number << N - number.bit_length()
        cumulative_sum += number
        numbers.append(cumulative_sum)
        bitstream = bitstream >> N
    return numbers

Но это очень медленно для огромных списков ввода

start = 5000
size = int(4e6)
numbers = [i for i in range(start, start + size + 1)]

Я начал тест с входным списком размером четыре миллиона, все элементы были последовательными, на кодирование и декодирование на моем ноутбуке ушло 13 минут.

Я думал о распараллеливании того, что у меня уже есть, но я не думаю, что есть где-нибудь, в чем я действительно смогу работать.

Не могли бы вы помочь мне просмотреть мой алгоритм и посмотреть, есть ли способы улучшить или переработать его, чтобы ускорить мои методы сжатия и распаковки?

Или, возможно, это может быть предел гамма-кодирования Элиаса в Python …

Я также заметил проблему с моей: кажется, что память не освобождается после ее завершения. tracemalloc.get_traced_memory() после запуска он показывает использование памяти выше, чем закодированное int.

Гамма-кодирование Элиаса кодирует такие числа

+--------+---------------+
| Number |  γ encoding   |
+--------+---------------+
|      1 |             1 |
|      2 |          0 10 |
|      3 |          0 11 |
|      4 |        00 100 |
|      5 |        00 101 |
|      6 |        00 110 |
|      7 |        00 111 |
|    ... |               |
|     15 |      000 1111 |
|     16 |   0000 1 0000 |
|    ... |               |
|     31 |   0000 1 1111 |
|     32 | 00000 10 0000 |

Начальные нули сообщают декодеру, сколько из следующих битов являются частью текущего числа. Таким образом, числа входного списка объединяются в поток битов и могут быть однозначно декодированы обратно в тот же список чисел. [1, 4, 3] кодирует в 1 00100 011.

Я храню битовый поток в обратном порядке внутри питона int, обратное, чтобы не было проблем с начальными нулями контейнера int.

Несмотря на то, что битовые последовательности в таблице выглядят большими из-за ведущих нулей, мое сжатие кодирует только разницу между текущим элементом списка и предыдущим (дельта-кодирование), так что в большинстве случаев число для кодирования и объединения намного меньше, чем фактический элемент списка.

Мои методы намеренно работают только с отсортированными списками.

Довольно странно кодировать как число вместо (битовой) строки. И то, как вы это делаете, требует квадратичного времени в размере вашего списка (потому что оба + и << в bitstream += encoded_number << bitstream_length линейны). Вы можете сделать это за линейное время, построив строку, а затем превратив ее в число в конце.

Это займет всего несколько секунд для вашего большого теста (здесь для краткости удалите строку документации):

def compress(numbers: List[int]) -> int:
    bitstream = []
    previous_number = 0
    for number in numbers:
        delta = number - previous_number # store deltas, much smaller numbers
        N = delta.bit_length() - 1
        encoded_number = bin(delta)[:1:-1] + '0' * N
        bitstream.append(encoded_number)
        previous_number = number
    bitstream.reverse()
    return int(''.join(bitstream), 2)

Соответственно, для декомпрессии я бы сначала превратил число в строку, а затем работал бы с этим.

Оптимизированная версия должна занять меньше секунды:

def compress(numbers: List[int]) -> int:
    bitstream = []
    append = bitstream.append
    memo = {}
    lookup = memo.get
    previous_number = 0
    for number in numbers:
        delta = number - previous_number # store deltas, much smaller numbers
        if not (encoded_number := lookup(delta)):
            N = delta.bit_length() - 1
            memo[delta] = encoded_number = bin(delta)[:1:-1] + '0' * N
        append(encoded_number)
        previous_number = number
    bitstream.reverse()
    return int(''.join(bitstream), 2)