Convert a bytes iterable to an iterable of str, where each value is a line

Question

I have an iterable of bytes, such as

bytes_iter = (
    b'col_1,',
    b'c',
    b'ol_2\n1',
    b',"val',
    b'ue"\n',
)

(but typically this would not be hard coded or available all at once, but supplied from a generator say) and I want to convert this to an iterable of str lines, where line breaks are unknown up front, but could be any of \r, \n or \r\n. So in this case would be:

lines_iter = (
    'col_1,col_2',
    '1,"value"',
)

(but again, just as an iterable, not so it's all in memory at once).

How can I do this?

Context: my aim is to then pass the iterable of str lines to csv.reader (that I think needs whole lines?), but I'm interested in this answer just in general.

Answer 1

Use the io module to do most of the work for you:

class ReadableIterator(io.IOBase):
    def __init__(self, it):
        self.it = iter(it)
    def read(self, n):
        # ignore argument, nobody actually cares
        # note that it is *critical* that we suppress the `StopIteration` here
        return next(self.it, b'')
    def readable(self):
        return True

then just call io.TextIOWrapper(ReadableIterator(some_iterable_of_bytes)).

Answer 2

I used yield and re.finditer.

The yield expression is used when defining a generator function or an asynchronous generator function and thus can only be used in the body of a function definition. Using a yield expression in a function’s body causes that function to be a generator function

Return an iterator yielding match objects over all non-overlapping matches for the RE pattern in string. The string is scanned left-to-right, and matches are returned in the order found. Empty matches are included in the result.
If there are no groups, return a list of strings matching the whole pattern. If there is exactly one group, return a list of strings matching that group. If multiple groups are present, return a list of tuples of strings matching the groups. Non-capturing groups do not affect the form of the result.

The regular expression ([^\r\n]*)(\r\n|\r|\n)? can be divided into two parts to match (that is, two groups). The first group matches the data without \r and \n, and the second group matches \r, \n or \r\n.

import re
find_rule = re.compile("([^\r\n]*)(\r\n|\r|\n)?")


def converter(byte_data):
    left_d = ""
    for d in byte_data:
        # Used to save the previous match result in the `for` loop
        prev_result = None
        # Concatenate the last part of the previous data with the current data,
        # used to deal with the case of `\r\n` being separated.
        d = left_d + d.decode()
        left_d = ""
        # Using `find_rule.finditer` the last value("") will be invalid
        for match_result in find_rule.finditer(d):
            i = match_result.group()
            if not i:
                # The program comes to this point, indicating that i == "", which is the last matching value
                left_d, prev_result = prev_result.group(), None
                continue
            if prev_result:
                if prev_result.group(2) is None:
                    # The program goes here, represented as the last valid value matched
                    left_d = prev_result.group()
                else:
                    # Returns the previous matched value
                    yield prev_result.group()
            # Save the current match result
            prev_result = match_result

    else:
        yield left_d


for i in (converter(iter((
        b'col_1,\r',
        b'\nc',
        b'ol_2\n1',
        b'\n,"val;\r',
        b'ue"\n')))
):
    print(repr(i))

Output:

'col_1,\r\n'
'col_2\n'
'1\n'
',"val;\r'
'ue"\n'

Answer 3

Maybe I'm missing something important (or subtle) because some of the upvoted answers seem a little more exotic than this, but I think you can decode and chain the bytes and use itertools.groupby to get a generator of strings:

from itertools import groupby, chain

bytes_iter = (
    b'col_1,',
    b'c',
    b'ol_2\n',
    b'1,"val;',
    b'ue"\n'
)

def make_strings(G):
    strings = chain.from_iterable(map(bytes.decode, G))
    for k, g in groupby(strings, key=lambda c: c not in '\n\r'):
        if k:
            yield ''.join(g)                            

list(make_strings(bytes_iter))
# ['col_1,col_2', '1,"val;ue"']

Answer 4

Putting @o11c's and @user2357112 supports Monica's contributions together:

import codecs
import csv
import io

def yield_bytes():
    chunks = [
        b'col_1,',
        b'c',
        b'ol_2\n1',
        b',"val',
        b'ue"\n',
        b'Hello,'
        b'\xe4\xb8',
        b'\x96',
        b'\xe7',
        b'\x95\x8c\n'
        b'\n'
    ]

    for chunk in chunks:
        yield(chunk)

decoder = codecs.getincrementaldecoder('utf-8')()

def yield_encoded_bytes():
    s = None
    for bytes in yield_bytes():
        s = decoder.decode(bytes, final=False)
        if s:
            yield s.encode('utf-8')

class ReadableIterator(io.IOBase):
    def __init__(self, it):
        self.it = iter(it)
    def read(self, n):
        # ignore argument, nobody actually cares
        # note that it is *critical* that we suppress the `StopIteration` here
        return next(self.it, b'')
    def readable(self):
        return True

f = io.TextIOWrapper(ReadableIterator(yield_encoded_bytes()))

for row in csv.reader(f):
    print(row)

and I get:

['col_1', 'col_2']
['1', 'value']
['Hello', '世界']
[]