LLM_DataGen / utils.py
Quentin Lhoest
initial commit
4f83ec0
import io
import logging
logger = logging.getLogger(__name__)
class StringIteratorIO(io.TextIOBase):
"""From: https://stackoverflow.com/a/12604375"""
def __init__(self, iter):
self._iter = iter
self._left = ''
def readable(self):
return True
def _read1(self, n=None):
while not self._left:
try:
self._left = next(self._iter)
except StopIteration:
break
ret = self._left[:n]
self._left = self._left[len(ret):]
return ret
def read(self, n=None):
buf = []
if n is None or n < 0:
while True:
m = self._read1()
if not m:
break
buf.append(m)
else:
while n > 0:
m = self._read1(n)
if not m:
break
n -= len(m)
buf.append(m)
return ''.join(buf)
def readline(self):
buf = []
while True:
i = self._left.find('\n')
if i == -1:
buf.append(self._left)
try:
self._left = next(self._iter)
except StopIteration:
self._left = ''
break
else:
buf.append(self._left[:i+1])
self._left = self._left[i+1:]
break
return ''.join(buf)