Модуль codecs в Python, реестр кодеков.

Реестр кодеков и базовые классы.

Модуль codecs определяет базовые классы для стандартных кодеков Python (кодировщик и декодировщик) и предоставляет доступ к внутреннему реестру кодеков Python, который управляет процессом поиска кодека и обработки ошибок.

Большинство стандартных кодеков - это текстовые кодировки, которые кодируют текст в байты, но есть также кодеки, которые кодируют текст в текст и байты в байты. В Python уже имеется большое количество стандартных кодеков и маловероятно, что приложению потребуется определять собственный кодировщик или декодировщик. Но иногда необходимо сделать свой кодек, для этих целей модуль codecs, предоставляет несколько базовых классов, чтобы упростить процесс.

Допустим надо сделать кодек, который будет инвертировать заглавные буквы в строчные, а строчные - в заглавные. Можно пойти простым путем и определить функцию кодирования, которая будет выполнять это преобразование для входной строки

import string

def invert(text):
    """Возвращает новую строку с преобразованными 
       заглавными буквами в строчные, а строчные - в заглавные
    """
    return ''.join(
        c.upper() if c in string.ascii_lowercase
        else c.lower() if c in string.ascii_uppercase
        else c
        for c in text
    )

>>> invert('Hello World')
'hELLO wORLD'
>>> invert('hELLO wORLD')
'Hello World'

Такая реализация неэффективна, особенно для очень больших текстовых строк. Модуль codecs включает в себя некоторые вспомогательные функции для создания кодеков, основанных на карте символов. Карта символов состоит из двух словарей. Карта преобразует символьные значения из входной строки в байтовые значения на выходе, а карта декодирования - в обратном направлении. Сначала создадим карту декодирования, а затем с помощью функции make_encoding_map() преобразуйте ее в карту кодирования.

# encoding_decoding_map.py

import codecs
import string

# сопоставить символы самим себе
decoding_map = codecs.make_identity_dict(range(256))

# список пар порядковых номеров соответствий
# строчной букв - заглавной
pairs = list(zip(
    [ord(c) for c in string.ascii_lowercase],
    [ord(c) for c in string.ascii_uppercase],
))

# теперь изменим карту 'decoding_map' так чтобы номера 
# строчных букв соответствовали заглавной и наоборот
decoding_map.update({
    upper: lower
    for (lower, upper)
    in pairs
})
decoding_map.update({
    lower: upper
    for (lower, upper)
    in pairs
})

# Создадим отдельную карту кодирования 
encoding_map = codecs.make_encoding_map(decoding_map)


>>> codecs.charmap_encode('Hello World', 'strict', encoding_map)
# (b'hELLO wORLD', 11)
>>> codecs.charmap_encode(b'hELLO wORLD', 'strict', decoding_map)
('Hello World', 11)

В данном случае карты кодирования и декодирования символов одинаковы, это может быть не всегда так. Функция make_encoding_map() обнаруживает ситуации, когда более одного входного символа кодируется в один и тот же выходной байт и заменяет значение кодирования на None, чтобы пометить кодировку как неопределенную.

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

Для создания полноценного кодека необходимо установить несколько дополнительных классов и зарегистрировать кодировщик и декодировщик. Функция codecs.register() добавляет функцию поиска в реестр кодеков, чтобы ее можно было найти. Функция поиска должна принимать один строковый аргумент с именем кодировки и возвращать объект codecs.CodecInfo(), если кодировка найдена или None если нет.

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

from encoding_decoding_map import encoding_map, decoding_map
import codecs

class InvertCodec(codecs.Codec):
    "кодирование/декодирование без сохранения состояния"

    def encode(self, input, errors='strict'):
        return codecs.charmap_encode(input, errors, encoding_map)

    def decode(self, input, errors='strict'):
        return codecs.charmap_decode(input, errors, decoding_map)

class InvertIncrementalEncoder(codecs.IncrementalEncoder):
    def encode(self, input, final=False):
        data, nbytes = codecs.charmap_encode(input,
                                             self.errors, encoding_map)
        return data

class InvertIncrementalDecoder(codecs.IncrementalDecoder):
    def decode(self, input, final=False):
        data, nbytes = codecs.charmap_decode(input,
                                             self.errors, decoding_map)
        return data

class InvertStreamReader(InvertCodec, codecs.StreamReader):
    pass

class InvertStreamWriter(InvertCodec, codecs.StreamWriter):
    pass


def find_invert(encoding):
    """Return the codec for 'invert'.
    """
    if encoding == 'invert':
        return codecs.CodecInfo(
            name='invert',
            encode=InvertCodec().encode,
            decode=InvertCodec().decode,
            incrementalencoder=InvertIncrementalEncoder,
            incrementaldecoder=InvertIncrementalDecoder,
            streamreader=InvertStreamReader,
            streamwriter=InvertStreamWriter,
        )
    return None

codecs.register(find_invert)

Для кодирования/декодирования без сохранения состояния переопределяем Codec.encode() и Codec.decode() с новой реализацией, вызывая функции charmap_encode() и charmap_decode() соответственно. Каждый метод должен возвращать кортеж, содержащий преобразованные данные и количество использованных входных байтов или символов.

IncrementalEncoder и IncrementalDecoder служат в качестве базовых классов для дополнительных интерфейсов. Методы encode() и decode() инкрементных классов определены таким образом, что они возвращают только фактические преобразованные данные. Любая информация о буферизации сохраняется как внутреннее состояние. Созданный кодек invert не требует буферизации данных, т.к. использует взаимно однозначное отображение.

StreamReader и StreamWriter также нуждаются в методах encode() и decode () т. к. они должны возвращать то же значение, что и версия из кодека, для реализации может использоваться множественное наследование.

# Stateless encoder/decoder
    encoder = codecs.getencoder('invert')
    text = 'Hello Word'
    encoded_text, consumed = encoder(text)
    print('Encoded "{}" to "{}", consuming {} characters'.format(
        text, encoded_text, consumed))

    # Stream writer
    import io
    buffer = io.BytesIO()
    writer = codecs.getwriter('invert')(buffer)
    print('StreamWriter for io buffer: ')
    print('  writing "abcDEF"')
    writer.write('abcDEF')
    print('  buffer contents: ', buffer.getvalue())

    # Incremental decoder
    decoder_factory = codecs.getincrementaldecoder('invert')
    decoder = decoder_factory()
    decoded_text_parts = []
    for c in encoded_text:
        decoded_text_parts.append(
            decoder.decode(bytes([c]), final=False)
        )
    decoded_text_parts.append(decoder.decode(b'', final=True))
    decoded_text = ''.join(decoded_text_parts)
    print('IncrementalDecoder converted {!r} to {!r}'.format(
        encoded_text, decoded_text))

# Encoded "Hello Word" to "b'hELLO wORLD'", consuming 6 characters
# StreamWriter for io buffer:
#   writing "Hello Word"
#   buffer contents:  b'hELLO wORLD'
# IncrementalDecoder converted b'hELLO wORLD' to 'Hello Word'