blob: 1d32d1082b048ec3be509373b69adf3f14769c6a (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
# SPDX-License-Identifier: GPL-2.0+
import functools
import typing
ByteIterable = typing.Iterable[bytes]
def yield_lines(iterable: ByteIterable) -> ByteIterable:
"""Converts an arbitrary bytes iterable into an iterable that yields whole
lines. The final byte of each returned value (except possibly the last one)
is a newline or carriage return character. The concatenation of the input
iterable equals the concatenation of the output iterable."""
buff = b""
for data in iterable:
buff += data
parts = buff.splitlines(True)
buff = parts.pop()
yield from parts
if buff:
yield buff
def decompress_stream(iterable: ByteIterable, decompressor) -> ByteIterable:
"""Decompress an iterable of bytes using the given decompressor into
another (decompressed) iterable of bytes. The decompressor can be a
bz2.BZ2Decompressor or lzma.LZMADecompressor instance."""
yield from map(decompressor.decompress, iterable)
if hasattr(decompressor, "flush"):
yield decompressor.flush()
def yield_chunks(filelike, chunksize=65536) -> ByteIterable:
"""Read the given file in chunks of the given size. Returns an itrable
of contents. If the file is binary, it yields bytes, otherwise str."""
return iter(functools.partial(filelike.read, chunksize), b"")
|