import sys

from pypy.interpreter.baseobjspace import W_Root, BufferInterfaceNotFound
from pypy.interpreter.error import OperationError, oefmt
from pypy.interpreter.gateway import WrappedDefault, interp2app, unwrap_spec
from pypy.interpreter.typedef import (
    GetSetProperty, TypeDef, generic_new_descr, interp_attrproperty,
    interp_attrproperty_w)
from pypy.module._codecs import interp_codecs
from pypy.module._io.interp_iobase import W_IOBase, convert_size, trap_eintr
from pypy.module.posix.interp_posix import device_encoding
from rpython.rlib.rarithmetic import intmask, r_uint, r_ulonglong
from rpython.rlib.rbigint import rbigint
from rpython.rlib.rstring import StringBuilder
from rpython.rlib.rutf8 import (check_utf8, next_codepoint_pos,
                                codepoints_in_utf8, codepoints_in_utf8,
                                Utf8StringBuilder)
from rpython.rlib import rlocale, jit
from rpython.rlib.objectmodel import always_inline


STATE_ZERO, STATE_OK, STATE_DETACHED = range(3)

SEEN_CR   = 1
SEEN_LF   = 2
SEEN_CRLF = 4
SEEN_ALL  = SEEN_CR | SEEN_LF | SEEN_CRLF

_WINDOWS = sys.platform == 'win32'

def make_newlines_dict(space):
    return {
        SEEN_CR: space.newtext("\r", 1),
        SEEN_LF: space.newtext("\n", 1),
        SEEN_CRLF: space.newtext("\r\n", 2),
        SEEN_CR | SEEN_LF: space.newtuple(
            [space.newtext("\r", 1),
             space.newtext("\n", 1)]),
        SEEN_CR | SEEN_CRLF: space.newtuple(
            [space.newtext("\r", 1),
             space.newtext("\r\n", 2)]),
        SEEN_LF | SEEN_CRLF: space.newtuple(
            [space.newtext("\n", 1),
             space.newtext("\r\n", 2)]),
        SEEN_CR | SEEN_LF | SEEN_CRLF: space.newtuple(
            [space.newtext("\r", 1),
             space.newtext("\n", 1),
             space.newtext("\r\n", 2)]),
    }

def io_check_errors(space, w_errors):
    # Make sure w_errors is valid:
    # - no \0
    # - can be encoded into utf8, no surrogates
    # - if dev_mode, is a valid error handler
    errors = space.text0_w(w_errors)
    space.encode_unicode_object(w_errors, 'utf8', 'strict')
    if not space.sys.get_flag('dev_mode'):
        return
    interp_codecs.lookup_error(space, errors)

class W_IncrementalNewlineDecoder(W_Root):
    seennl = 0
    pendingcr = False
    w_decoder = None

    def __init__(self, space):
        pass

    @unwrap_spec(translate=int)
    def descr_init(self, space, w_decoder, translate, w_errors=None):
        self.w_decoder = w_decoder
        self.translate = translate
        if space.is_none(w_errors) or w_errors is None:
            w_errors = space.newtext("strict")
        if not space.isinstance_w(w_errors, space.w_text):
            raise oefmt(space.w_TypeError, "TextIOWrapper() argument 'errors' "
                "must be str or None, not %N", w_errors)
        io_check_errors(space, w_errors)
        self.w_errors = w_errors

        self.seennl = 0

    def newlines_get_w(self, space):
        return space.fromcache(make_newlines_dict).get(self.seennl, space.w_None)

    @unwrap_spec(final=int)
    def decode_w(self, space, w_input, final=False):
        if self.w_decoder is None:
            raise oefmt(space.w_ValueError,
                        "IncrementalNewlineDecoder.__init__ not called")

        # decode input (with the eventual \r from a previous pass)
        if not space.is_w(self.w_decoder, space.w_None):
            w_output = space.call_method(self.w_decoder, "decode",
                                         w_input, space.newbool(bool(final)))
        else:
            w_output = w_input

        if not space.isinstance_w(w_output, space.w_unicode):
            raise oefmt(space.w_TypeError,
                        "decoder should return a string result")

        output, output_len = space.utf8_len_w(w_output)
        w_result = self._decoder_w_unboxed(space, output, output_len, final)
        if w_result is None:
            return w_output
        return w_result

    def _decoder_w_unboxed(self, space, output, output_len, final):
        changed = False
        if self.pendingcr and (final or output_len):
            output = '\r' + output
            self.pendingcr = False
            output_len += 1
            changed = True

        # retain last \r even when not translating data:
        # then readline() is sure to get \r\n in one pass
        if not final and output_len > 0:
            last = len(output) - 1
            assert last >= 0
            if output[last] == '\r':
                output = output[:last]
                self.pendingcr = True
                output_len -= 1
                changed = True

        if output_len == 0:
            return space.newutf8("", 0)

        # Record which newlines are read and do newline translation if
        # desired, all in one pass.
        seennl = self.seennl

        rpos = output.find('\r')
        if rpos < 0:
            # If no \r, quick scan for a possible "\n" character.
            # (there's nothing else to be done, even when in translation mode)
            if not (seennl & SEEN_LF) and output.find('\n') >= 0:
                self.seennl |= SEEN_LF
            # Finished: we have scanned for newlines, and none of them
            # need translating.
            if not changed:
                return None # means we can reuse the w_output in the caller
        elif not self.translate:
            i = 0
            while i < len(output):
                if seennl == SEEN_ALL:
                    break
                c = output[i]
                i += 1
                if c == '\n':
                    seennl |= SEEN_LF
                elif c == '\r':
                    if i < len(output) and output[i] == '\n':
                        seennl |= SEEN_CRLF
                        i += 1
                    else:
                        seennl |= SEEN_CR
            if not changed:
                self.seennl |= seennl
                return None
        else:
            changed = True
            assert rpos >= 0
            # Translate!
            builder = StringBuilder(len(output))
            if output.find("\n", 0, rpos) >= 0:
                seennl |= SEEN_LF
            builder.append_slice(output, 0, rpos)
            i = rpos
            while i < len(output):
                c = output[i]
                i += 1
                if c == '\n':
                    seennl |= SEEN_LF
                elif c == '\r':
                    if i < len(output) and output[i] == '\n':
                        seennl |= SEEN_CRLF
                        i += 1
                        output_len -= 1
                    else:
                        seennl |= SEEN_CR
                    builder.append('\n')
                    continue
                builder.append(c)
            output = builder.build()
        self.seennl |= seennl
        return space.newutf8(output, output_len)

    def reset_w(self, space):
        self.seennl = 0
        self.pendingcr = False
        if self.w_decoder and not space.is_w(self.w_decoder, space.w_None):
            space.call_method(self.w_decoder, "reset")

    def getstate_u(self, space):
        if self.w_decoder and not space.is_w(self.w_decoder, space.w_None):
            w_state = space.call_method(self.w_decoder, "getstate")
            w_buffer, w_flag = space.unpackiterable(w_state, 2)
            flag = space.r_longlong_w(w_flag)
        else:
            w_buffer = space.newbytes("")
            flag = 0
        flag <<= 1
        if self.pendingcr:
            flag |= 1
        return w_buffer, flag

    def getstate_w(self, space):
        w_buffer, flag = self.getstate_u(space)
        return space.newtuple2(w_buffer, space.newint(flag))

    def setstate_w(self, space, w_state):
        w_buffer, w_flag = space.unpackiterable(w_state, 2)
        flag = space.r_longlong_w(w_flag)
        self.pendingcr = bool(flag & 1)
        flag >>= 1

        if self.w_decoder and not space.is_w(self.w_decoder, space.w_None):
            w_state = space.newtuple2(w_buffer, space.newint(flag))
            space.call_method(self.w_decoder, "setstate", w_state)

W_IncrementalNewlineDecoder.typedef = TypeDef(
    '_io.IncrementalNewlineDecoder',
    __new__ = generic_new_descr(W_IncrementalNewlineDecoder),
    __init__  = interp2app(W_IncrementalNewlineDecoder.descr_init),

    decode = interp2app(W_IncrementalNewlineDecoder.decode_w),
    reset = interp2app(W_IncrementalNewlineDecoder.reset_w),
    getstate = interp2app(W_IncrementalNewlineDecoder.getstate_w),
    setstate = interp2app(W_IncrementalNewlineDecoder.setstate_w),

    newlines = GetSetProperty(W_IncrementalNewlineDecoder.newlines_get_w),
)

class W_TextIOBase(W_IOBase):
    w_encoding = None

    def __init__(self, space):
        W_IOBase.__init__(self, space)

    def read_w(self, space, w_size=None):
        self._unsupportedoperation(space, "read")

    def readline_w(self, space, w_limit=None):
        self._unsupportedoperation(space, "readline")

    def write_w(self, space, w_data):
        self._unsupportedoperation(space, "write")

    def detach_w(self, space):
        self._unsupportedoperation(space, "detach")

    def errors_get_w(self, space):
        return space.w_None

    def newlines_get_w(self, space):
        return space.w_None

W_TextIOBase.typedef = TypeDef(
    '_io._TextIOBase', W_IOBase.typedef,
    __new__ = generic_new_descr(W_TextIOBase),

    read = interp2app(W_TextIOBase.read_w),
    readline = interp2app(W_TextIOBase.readline_w),
    write = interp2app(W_TextIOBase.write_w),
    detach = interp2app(W_TextIOBase.detach_w),
    encoding = interp_attrproperty_w("w_encoding", W_TextIOBase),
    newlines = GetSetProperty(W_TextIOBase.newlines_get_w),
    errors = GetSetProperty(W_TextIOBase.errors_get_w),
)


def _determine_encoding(space, encoding, w_buffer):
    if encoding == "locale":
        pass
    elif encoding is None:
        _maybe_warn_encoding(space)
    else:
        return space.newtext(encoding)

    # Try os.device_encoding(fileno) which is interp_posix.device_encoding
    try:
        w_fileno = space.call_method(w_buffer, 'fileno')
    except OperationError as e:
        from pypy.module._io.interp_io import Cache
        if not (e.match(space, space.w_AttributeError) or
                e.match(space, space.fromcache(Cache).w_unsupportedoperation)):
            raise
    else:
        try:
            w_encoding = device_encoding(space, space.int_w(w_fileno))
        except OverflowError:
            raise oefmt(space.w_OverflowError,
                        "Python int too large to convert to C int")
        else:
            if space.isinstance_w(w_encoding, space.w_unicode):
                return w_encoding
    if space.sys.get_flag('utf8_mode'):
        # if device_encoding returns None, CPython does
        # _Py_GetLocaleEncoding, which has this at the top
        return space.newtext("utf-8")

    # On legacy systems or darwin, try app-level
    # _bootlocale.getprefferedencoding(False)
    try:
        w_locale = space.call_method(space.builtin, '__import__',
                                     space.newtext('_bootlocale'))
        w_encoding = space.call_method(w_locale, 'getpreferredencoding',
                                       space.w_False)
    except OperationError as e:
        # getpreferredencoding() may also raise ImportError
        if not e.match(space, space.w_ImportError):
            raise
        return space.newtext('ascii')
    else:
        if space.isinstance_w(w_encoding, space.w_text):
            return w_encoding

    raise oefmt(space.w_IOError, "could not determine default encoding")

@unwrap_spec(stacklevel=int)
def text_encoding(space, w_encoding, stacklevel=2):
    """
    A helper function to choose the text encoding.

    When encoding is not None, just return it.
    Otherwise, return the default text encoding (i.e. "locale").

    This function emits an EncodingWarning if *encoding* is None and
    sys.flags.warn_default_encoding is non-zero.

    This can be used in APIs with an encoding=None parameter
    that pass it to TextIOWrapper or open.
    However, please consider using encoding="utf-8" for new APIs.
    """
    if space.is_none(w_encoding):
        _maybe_warn_encoding(space, stacklevel + 1)
        if space.sys.get_flag('utf8_mode') == 0:
            return space.newtext("locale")
        else:
            return space.newtext("utf-8")
    return w_encoding

def _maybe_warn_encoding(space, stacklevel=1):
    if space.sys.get_flag('warn_default_encoding') != 0:
        space.warn(space.newtext("'encoding' argument not specified."),
                   space.w_EncodingWarning, stacklevel)

class PositionCookie(object):
    def __init__(self, bigint):
        self.start_pos = bigint.ulonglongmask()
        bigint = bigint.rshift(r_ulonglong.BITS)
        x = intmask(bigint.uintmask())
        assert x >= 0
        self.dec_flags = x
        bigint = bigint.rshift(r_uint.BITS)
        x = intmask(bigint.uintmask())
        assert x >= 0
        self.bytes_to_feed = x
        bigint = bigint.rshift(r_uint.BITS)
        x = intmask(bigint.uintmask())
        assert x >= 0
        self.chars_to_skip = x
        bigint = bigint.rshift(r_uint.BITS)
        self.need_eof = bigint.tobool()

    def pack(self):
        # The meaning of a tell() cookie is: seek to position, set the
        # decoder flags to dec_flags, read bytes_to_feed bytes, feed them
        # into the decoder with need_eof as the EOF flag, then skip
        # chars_to_skip characters of the decoded result.  For most simple
        # decoders, tell() will often just give a byte offset in the file.
        rb = rbigint.fromrarith_int

        res = rb(self.start_pos)
        bits = r_ulonglong.BITS
        res = res.or_(rb(r_uint(self.dec_flags)).lshift(bits))
        bits += r_uint.BITS
        res = res.or_(rb(r_uint(self.bytes_to_feed)).lshift(bits))
        bits += r_uint.BITS
        res = res.or_(rb(r_uint(self.chars_to_skip)).lshift(bits))
        bits += r_uint.BITS
        return res.or_(rb(r_uint(self.need_eof)).lshift(bits))

    def __repr__(self):
        return "coookie w/start_pos %d dec_flags %d bytes_to_feed %d chars_to_skip %d need_eof %d" %(self.start_pos, self.dec_flags, self.bytes_to_feed, self.chars_to_skip, self.need_eof)

class PositionSnapshot:
    _immutable_fields_ = ["flags", "input"]
    def __init__(self, flags, input):
        self.flags = flags
        self.input = input


class DecodeBuffer(object):
    def __init__(self, text=None, ulen=-1):
        # self.text is a valid utf-8 string
        if text is not None:
            assert ulen >= 0
        self.text = text
        self.pos = 0   # in utf8
        self.upos = 0  # decoded_chars_used in CPython, in codepoints
        self.ulen = ulen

    def set(self, space, w_decoded):
        # set_chars_decoded in CPython
        check_decoded(space, w_decoded)
        self.text, self.ulen = space.utf8_len_w(w_decoded)
        self.pos = 0
        self.upos = 0

    def reset(self):
        self.text = None
        self.pos = 0
        self.upos = 0
        self.ulen = -1

    def get_chars(self, size):
        """ returns a tuple (utf8, lgt) """
        # get_decoded_chars in CPython
        if self.text is None or size == 0:
            return "", 0

        lgt = self.ulen
        available = lgt - self.upos
        if size < 0 or size > available:
            size = available
        assert size >= 0

        if self.pos > 0 or size < available:
            start = self.pos
            pos = start
            for i in range(size):
                pos = next_codepoint_pos(self.text, pos)
                self.upos += 1
            assert start >= 0
            assert pos >= 0
            chars = self.text[start:pos]
            self.pos = pos
        else:
            chars = self.text
            self.pos = len(self.text)
            self.upos = lgt
            size = lgt

        return chars, size

    def has_data(self):
        return (self.text is not None and not self.exhausted())

    def exhausted(self):
        return self.pos >= len(self.text)

    def find_newline_universal(self, limit):
        # Universal newline search. Find any of \r, \r\n, \n
        # The decoder ensures that \r\n are not split in two pieces
        if limit < 0:
            limit = sys.maxint
        scanned = 0
        while scanned < limit:
            if self.exhausted():
                return False
            ch = self.text[self.pos]
            self._advance_codepoint()
            scanned += 1
            if ch == '\n':
                return True
            if ch == '\r':
                if scanned >= limit:
                    return False
                if self.exhausted():
                    return True
                ch = self.text[self.pos]
                if ch == '\n':
                    self.pos += 1
                    self.upos += 1
                    return True
                else:
                    return True
        return False

    def find_crlf(self, limit):
        if limit < 0:
            limit = sys.maxint
        scanned = 0
        while scanned < limit:
            if self.exhausted():
                return False
            ch = self.text[self.pos]
            scanned += 1
            if ch == '\r':
                self.pos += 1
                self.upos += 1
                if scanned >= limit:
                    return False
                if self.exhausted():
                    # This is the tricky case: we found a \r right at the end,
                    # un-consume it
                    self.pos -= 1
                    self.upos -= 1
                    return False
                if self.text[self.pos] == '\n':
                    self.pos += 1
                    self.upos += 1
                    return True
            else:
                self._advance_codepoint()
        return False

    def find_char(self, marker, limit):
        # only works for ascii markers!
        assert 0 <= ord(marker) < 128
        # ascii fast path
        if self.ulen == len(self.text):
            end = len(self.text)
            if limit >= 0:
                end = min(end, self.pos + limit)
            pos = self.pos
            assert pos >= 0
            assert end >= 0
            pos = self.text.find(marker, pos, end)
            if pos >= 0:
                self.pos = self.upos = pos + 1
                return True
            else:
                self.pos = self.upos = end
                return False

        if limit < 0:
            limit = sys.maxint
        # XXX it might be better to search for the marker quickly, then compute
        # the new upos afterwards.
        scanned = 0
        while scanned < limit:
            # don't use next_char here, since that computes a slice etc
            if self.exhausted():
                return False
            # this is never true if self.text[pos] is part of a larger char
            found = self.text[self.pos] == marker
            if found:
                self.pos += 1
                self.upos += 1
                return True
            self._advance_codepoint()
            scanned += 1
        return False

    @always_inline
    def _advance_codepoint(self):
        # must only be called after checking self.exhausted()!
        self.pos = next_codepoint_pos(self.text, self.pos)
        self.upos += 1


def check_decoded(space, w_decoded):
    if not space.isinstance_w(w_decoded, space.w_unicode):
        msg = "decoder should return a string result, not '%T'"
        raise oefmt(space.w_TypeError, msg, w_decoded)
    return w_decoded

def unwrap_newline(space, w_newline):
    if space.is_none(w_newline):
        newline = None
    else:
        newline = space.utf8_w(w_newline)
        if newline and newline not in ('\n', '\r\n', '\r'):
            raise oefmt(space.w_ValueError,
                        "illegal newline value: %R", w_newline)
    return newline

class W_TextIOWrapper(W_TextIOBase):
    def __init__(self, space):
        W_TextIOBase.__init__(self, space)
        self.state = STATE_ZERO
        self.w_encoder = None
        self.w_decoder = None

        self.decoded = DecodeBuffer()
        self.pending_bytes = None   # list of bytes objects waiting to be
                                    # written, or NULL
        self.pending_bytes_count = 0
        self.chunk_size = 8192
        self.b2cratio = 0.0

        self.readuniversal = False
        self.readtranslate = False
        self.readnl = None

        self.encodefunc = None # Specialized encoding func (see below)
        self.encoding_start_of_stream = False # Whether or not it's the start
                                              # of the stream
        self.snapshot = None

    @unwrap_spec(encoding="text0_or_none", line_buffering=int, write_through=int)
    def descr_init(self, space, w_buffer, encoding=None,
                   w_errors=None, w_newline=None, line_buffering=0,
                   write_through=0):
        self.state = STATE_ZERO
        self.w_buffer = w_buffer
        self.w_encoding = w_encoding = _determine_encoding(space, encoding, w_buffer)

        if space.is_none(w_errors) or w_errors is None:
            w_errors = space.newtext("strict")
        if not space.isinstance_w(w_errors, space.w_text):
            raise oefmt(space.w_TypeError, "TextIOWrapper() argument 'errors' "
                "must be str or None, not %N", w_errors)
        io_check_errors(space, w_errors)
        self.w_errors = w_errors

        newline = unwrap_newline(space, w_newline)

        self.line_buffering = bool(line_buffering)
        self.write_through = bool(write_through)

        self._set_newline(newline)

        self._set_encoder_decoder(w_encoding, w_errors)

        self.seekable = space.is_true(space.call_method(w_buffer, "seekable"))
        self.telling = self.seekable

        self.has_read1 = space.findattr(w_buffer, space.newtext("read1"))

        self._fix_encoder_state()

        self.state = STATE_OK

    def _set_newline(self, newline):
        self.readuniversal = not newline # null or empty
        self.readtranslate = newline is None
        self.readnl = newline

        self.writetranslate = (newline != '')
        if not self.readuniversal:
            self.writenl = self.readnl
            if self.writenl == '\n':
                self.writenl = None
        elif _WINDOWS:
            self.writenl = "\r\n"
        else:
            self.writenl = None

    def _set_encoder_decoder(self, w_encoding, w_errors):
        space = self.space
        w_codec = interp_codecs.lookup_codec(space,
                                             space.text_w(w_encoding))
        if not space.is_true(space.getattr(w_codec,
                                           space.newtext('_is_text_encoding'))):
            msg = ("%R is not a text encoding; "
                   "use codecs.open() to handle arbitrary codecs")
            raise oefmt(space.w_LookupError, msg, w_encoding)

        # build the decoder object
        if space.is_true(space.call_method(self.w_buffer, "readable")):
            self.w_decoder = space.call_method(w_codec,
                                               "incrementaldecoder", w_errors)
            if self.readuniversal:
                self.w_decoder = space.call_function(
                    space.gettypeobject(W_IncrementalNewlineDecoder.typedef),
                    self.w_decoder, space.newbool(self.readtranslate))

        # build the encoder object
        if space.is_true(space.call_method(self.w_buffer, "writable")):
            self.w_encoder = space.call_method(w_codec,
                                               "incrementalencoder", w_errors)

    def _fix_encoder_state(self):
        space = self.space
        self.encoding_start_of_stream = False
        if self.seekable and self.w_encoder:
            self.encoding_start_of_stream = True
            w_cookie = space.call_method(self.w_buffer, "tell")
            if not space.eq_w(w_cookie, space.newint(0)):
                self.encoding_start_of_stream = False
                space.call_method(self.w_encoder, "setstate", space.newint(0))

    def reconfigure(self, space, __args__):
        """
        Reconfigure the text stream with new parameters.

        This also does an implicit stream flush.
        """
        # XXX quite annoying, kwonly args can't easily support unwrapped None
        # as the default, do our own argument parsing
        args_w, kwargs_w = __args__.unpack()
        if args_w:
            raise oefmt(space.w_TypeError, "reconfigure() takes no positional arguments")
        # for all arguments passing w_None means "keep value", with two exceptions:
        # 1) if encoding is given but not errors, set errors to strict
        # 2) newline=None means universal newline support
        w_encoding = kwargs_w.pop("encoding", space.w_None)
        w_errors = kwargs_w.pop("errors", space.w_None)
        w_newline = kwargs_w.pop("newline", None)
        if (not space.is_none(w_encoding) and
            not space.isinstance_w(w_encoding, space.w_text)):
            raise oefmt(space.w_TypeError,
                "reconfigure argument 'encoding' must be str or None, not %T",
                 w_encoding)
        if (not space.is_none(w_errors) and
            not space.isinstance_w(w_errors, space.w_text)):
            raise oefmt(space.w_TypeError,
                "reconfigure argument 'errors' must be str or None, not %T",
                 w_errors)
        if (not space.is_none(w_newline) and
            not space.isinstance_w(w_newline, space.w_text)):
            raise oefmt(space.w_TypeError,
                "reconfigure argument 'newline' must be str or None, not %T",
                 w_newline)
        w_line_buffering = kwargs_w.pop("line_buffering", space.w_None)
        w_write_through = kwargs_w.pop("write_through", space.w_None)
        if kwargs_w:
            key, w_value = kwargs_w.popitem()
            raise oefmt(space.w_TypeError, "%8 is an invalid keyword argument for reconfigure()", key)

        if self.decoded.text is not None:
            if (not space.is_none(w_encoding) or
                    not space.is_none(w_errors) or
                    w_newline is not None):
                self._unsupportedoperation(
                    space, "It is not possible to set the encoding "
                           "or newline of stream after the first read")


        newline = None
        if w_newline is not None:
            newline = unwrap_newline(space, w_newline)

        line_buffering = self.line_buffering
        if not space.is_none(w_line_buffering):
            line_buffering = bool(space.int_w(w_line_buffering))
        write_through = self.write_through
        if not space.is_none(w_write_through):
            write_through = bool(space.int_w(w_write_through))

        space.call_method(self, "flush")
        if w_newline is not None:
            self._set_newline(newline)
        if not space.is_none(w_errors):
            io_check_errors(space, w_errors)
        # if encoding is specified but not errors, set errors to strict
        if not space.is_none(w_encoding):
            if space.is_none(w_errors):
                w_errors = space.newtext("strict")
        if not space.is_none(w_encoding) or not space.is_none(w_errors) or (w_newline is not None and self.readuniversal):
            if space.is_none(w_encoding):
                w_encoding = self.w_encoding
            elif space.eq_w(w_encoding, space.newtext("locale")):
                w_encoding = space.newtext("utf-8")
            if space.is_none(w_errors):
                w_errors = self.w_errors
            # NB we also need to call _set_encoder_decoder if the newline
            # changed to readuniversal, to get newline translation
            self._set_encoder_decoder(w_encoding, w_errors)
            self.w_encoding = w_encoding
            self.w_errors = w_errors

        self.line_buffering = line_buffering
        self.write_through = write_through

        self._fix_encoder_state()
        self.b2cratio = 0.0

    def _check_init(self, space):
        if self.state == STATE_ZERO:
            raise oefmt(space.w_ValueError,
                        "I/O operation on uninitialized object")

    def _check_attached(self, space):
        if jit.promote(self.state) == STATE_DETACHED:
            raise oefmt(space.w_ValueError,
                        "underlying buffer has been detached")
        self._check_init(space)

    def _check_closed(self, space, message=None):
        self._check_init(space)
        W_TextIOBase._check_closed(self, space, message)

    def __w_attr_repr(self, space, name):
        w_attr = space.findattr(self, space.newtext(name))
        if w_attr is None:
            return space.newtext("")
        return space.mod(space.newtext("%s=%%r " % name), w_attr)

    def descr_repr(self, space):
        self._check_init(space)
        w_args = space.newtuple([self.__w_attr_repr(space, 'name'),
                                 self.__w_attr_repr(space, 'mode'),
                                 self.w_encoding])
        return space.mod(
            space.newtext("<_io.TextIOWrapper %s%sencoding=%r>"), w_args
        )

    def readable_w(self, space):
        self._check_attached(space)
        return space.call_method(self.w_buffer, "readable")

    def writable_w(self, space):
        self._check_attached(space)
        return space.call_method(self.w_buffer, "writable")

    def seekable_w(self, space):
        self._check_attached(space)
        return space.call_method(self.w_buffer, "seekable")

    def isatty_w(self, space):
        self._check_attached(space)
        return space.call_method(self.w_buffer, "isatty")

    def fileno_w(self, space):
        self._check_attached(space)
        return space.call_method(self.w_buffer, "fileno")

    def closed_get_w(self, space):
        self._check_attached(space)
        return space.getattr(self.w_buffer, space.newtext("closed"))

    def newlines_get_w(self, space):
        self._check_attached(space)
        if self.w_decoder is None:
            return space.w_None
        return space.findattr(self.w_decoder, space.newtext("newlines"))

    def name_get_w(self, space):
        self._check_attached(space)
        return space.getattr(self.w_buffer, space.newtext("name"))

    def flush_w(self, space):
        self._check_attached(space)
        self._check_closed(space)
        self.telling = self.seekable
        self._writeflush(space)
        space.call_method(self.w_buffer, "flush")

    @unwrap_spec(w_pos = WrappedDefault(None))
    def truncate_w(self, space, w_pos=None):
        self._check_attached(space)

        space.call_method(self, "flush")
        return space.call_method(self.w_buffer, "truncate", w_pos)

    def close_w(self, space):
        self._check_attached(space)
        if space.is_true(space.getattr(self.w_buffer,
                                       space.newtext("closed"))):
            return
        try:
            space.call_method(self, "flush")
        except OperationError as e:
            try:
                ret = space.call_method(self.w_buffer, "close")
            except OperationError as e2:
                e2.chain_exceptions(space, e)
            raise
        else:
            ret = space.call_method(self.w_buffer, "close")
        self.maybe_unregister_rpython_finalizer_io(space)
        return ret

    def _dealloc_warn_w(self, space, w_source):
        space.call_method(self.w_buffer, "_dealloc_warn", w_source)

    # _____________________________________________________________
    # read methods

    def _read_chunk(self, space, size_hint):
        """Read and decode the next chunk of data from the BufferedReader.
        The return value is True unless EOF was reached.  The decoded string
        is placed in self.decoded (replacing its previous value).
        The entire input chunk is sent to the decoder, though some of it may
        remain buffered in the decoder, yet to be converted."""

        if not self.w_decoder:
            self._unsupportedoperation(space, "not readable")

        if self.telling:
            # To prepare for tell(), we need to snapshot a point in the file
            # where the decoder's input buffer is empty.
            w_decoder = self.w_decoder
            # fast path for the common case of decoder being
            # W_IncrementalNewlineDecoder. avoids (un)wrapping the tuple too
            if type(w_decoder) is W_IncrementalNewlineDecoder:
                w_dec_buffer, dec_flags = w_decoder.getstate_u(space)
            else:
                w_state = space.call_method(self.w_decoder, "getstate")
                if (not space.isinstance_w(w_state, space.w_tuple)
                        or space.len_w(w_state) != 2):
                    raise oefmt(space.w_TypeError, "illegal decoder state")
                # Given this, we know there was a valid snapshot point
                # len(dec_buffer) bytes ago with decoder state (b'', dec_flags).
                w_dec_buffer, w_dec_flags = space.unpackiterable(w_state, 2)
                if not space.isinstance_w(w_dec_buffer, space.w_bytes):
                    msg = ("illegal decoder state: the first value should be a "
                        "bytes object not '%T'")
                    raise oefmt(space.w_TypeError, msg, w_dec_buffer)
                dec_flags = space.int_w(w_dec_flags)
            dec_buffer = space.bytes_w(w_dec_buffer)
        else:
            dec_buffer = None
            dec_flags = 0

        # Read a chunk, decode it, and put the result in self.decoded
        if size_hint > 0:
            size_hint = int(max(self.b2cratio, 1.0) * float(size_hint))
        chunk_size = max(self.chunk_size, size_hint)
        func_name = "read1" if self.has_read1 else "read"
        w_input = space.call_method(self.w_buffer, func_name,
                                    space.newint(chunk_size))

        try:
            input_buf = w_input.buffer_w(space, space.BUF_SIMPLE)
        except BufferInterfaceNotFound:
            msg = ("underlying %s() should have returned a bytes-like "
                   "object, not '%T'")
            raise oefmt(space.w_TypeError, msg, func_name, w_input)

        nbytes = input_buf.getlength()
        eof = nbytes == 0
        w_decoder = self.w_decoder
        if type(w_decoder) is W_IncrementalNewlineDecoder:
            w_decoded = w_decoder.decode_w(space, w_input, eof)
        else:
            w_decoded = space.call_method(w_decoder, "decode",
                                          w_input, space.newbool(eof))
        self.decoded.set(space, w_decoded)
        nchars = space.len_w(w_decoded)
        if nchars > 0:
            eof = False
            self.b2cratio = float(nbytes) / float(nchars)
        else:
            self.b2cratio = 0.0

        if self.telling:
            # At the snapshot point, len(dec_buffer) bytes before the read,
            # the next input to be decoded is dec_buffer + input_chunk.
            next_input = dec_buffer + input_buf.as_str()
            self.snapshot = PositionSnapshot(dec_flags, next_input)

        return not eof

    def _ensure_data(self, space, size_hint):
        while not self.decoded.has_data():
            try:
                if not self._read_chunk(space, size_hint):
                    self.decoded.reset()
                    self.snapshot = None
                    return False
            except OperationError as e:
                if trap_eintr(space, e):
                    continue
                raise
        return True

    def next_w(self, space):
        self._check_attached(space)
        self.telling = False
        try:
            return W_TextIOBase.next_w(self, space)
        except OperationError as e:
            if e.match(space, space.w_StopIteration):
                self.telling = self.seekable
            raise

    def read_w(self, space, w_size=None):
        self._check_attached(space)
        self._check_closed(space)
        if not self.w_decoder:
            self._unsupportedoperation(space, "not readable")

        size = convert_size(space, w_size)
        self._writeflush(space)

        if size < 0:
            return self._read_all(space)
        else:
            return self._read(space, size)

    def _read_all(self, space):
        w_bytes = space.call_method(self.w_buffer, "read")
        w_decoded = space.call_method(self.w_decoder, "decode", w_bytes, space.w_True)
        check_decoded(space, w_decoded)
        w_result = space.newutf8(*self.decoded.get_chars(-1))
        w_final = space.add(w_result, w_decoded)
        if self.snapshot:
            # CPython GH-35928
            self.decoded.reset()
            self.snapshot = None
        return w_final

    def _read(self, space, size):
        remaining = size
        builder = Utf8StringBuilder(size)

        # Keep reading chunks until we have n characters to return
        while remaining > 0:
            if not self._ensure_data(space, remaining):
                break
            data, size = self.decoded.get_chars(remaining)
            builder.append_utf8(data, size)
            remaining -= size

        return space.newutf8(builder.build(), builder.getlength())

    def _scan_line_ending(self, limit):
        if self.readtranslate:
            # Newlines are already translated, only search for \n
            return self.decoded.find_char('\n', limit)
        if self.readuniversal:
            return self.decoded.find_newline_universal(limit)
        else:
            # Non-universal mode.
            newline = self.readnl
            if newline == '\r\n':
                return self.decoded.find_crlf(limit)
            else:
                return self.decoded.find_char(newline[0], limit)

    def readline_w(self, space, w_limit=None):
        self._check_attached(space)
        self._check_closed(space)
        self._writeflush(space)
        limit = convert_size(space, w_limit)
        text, lgt = self._readline(space, limit)
        return space.newutf8(text, lgt)

    def _readline(self, space, limit):
        # This is a separate function so that readline_w() can be jitted.
        remnant = None
        remnant_ulen = -1
        builder = Utf8StringBuilder()
        while True:
            # First, get some data if necessary
            has_data = self._ensure_data(space, 0)
            if not has_data:
                # end of file
                if remnant:
                    builder.append_utf8(remnant, remnant_ulen)
                break

            if remnant:
                assert not self.readtranslate and self.readnl == '\r\n'
                assert self.decoded.pos == 0
                if remnant == '\r' and self.decoded.text[0] == '\n':
                    builder.append_utf8('\r\n', 2)
                    self.decoded.pos = 1
                    self.decoded.upos = 1
                    remnant = None
                    remnant_ulen = -1
                    break
                else:
                    builder.append_utf8(remnant, remnant_ulen)
                    remnant = None
                    remnant_ulen = -1
                    continue

            if limit >= 0:
                remaining = limit - builder.getlength()
                assert remaining >= 0
            else:
                remaining = -1
            start = self.decoded.pos
            ustart = self.decoded.upos
            assert start >= 0
            found = self._scan_line_ending(remaining)
            end_scan = self.decoded.pos
            uend_scan = self.decoded.upos
            if end_scan > start:
                builder.append_utf8_slice(self.decoded.text, start, end_scan, uend_scan - ustart)

            if found or (limit >= 0 and builder.getlength() >= limit):
                break

            # There may be some remaining chars we'll have to prepend to the
            # next chunk of data
            if not self.decoded.exhausted():
                remnant, remnant_ulen = self.decoded.get_chars(-1)
            # We have consumed the buffer
            self.decoded.reset()

        result = builder.build()
        lgt = builder.getlength()
        return (result, lgt)

    # _____________________________________________________________
    # write methods

    def write_w(self, space, w_text):
        self._check_attached(space)
        self._check_closed(space)

        if not self.w_encoder:
            self._unsupportedoperation(space, "not writable")

        if not space.isinstance_w(w_text, space.w_unicode):
            raise oefmt(space.w_TypeError,
                        "unicode argument expected, got '%T'", w_text)

        text, textlen = space.utf8_len_w(w_text)

        haslf = False
        if (self.writetranslate and self.writenl) or self.line_buffering:
            if text.find('\n') >= 0:
                haslf = True
        if haslf and self.writetranslate and self.writenl:
            w_text = space.call_method(w_text, "replace", space.newutf8('\n', 1),
                               space.newutf8(self.writenl, codepoints_in_utf8(self.writenl)))
            text = space.utf8_w(w_text)

        needflush = False
        text_needflush = False
        if self.write_through:
            text_needflush = True
        if self.line_buffering and (haslf or text.find('\r') >= 0):
            needflush = True

        # XXX What if we were just reading?
        if self.encodefunc:
            # XXX CPython has an optimization if
            # - isascii(w_text) and
            # - len(w_text) <= chunk_size and
            # - is_asciicompat_encoding(self.encodefunc)
            # then w_bytes = w_text
            w_bytes = self.encodefunc(space, w_text, self.errors)
            self.encoding_start_of_stream = False
        else:
            w_bytes = space.call_method(self.w_encoder, "encode", w_text)

        b = space.bytes_w(w_bytes)

        if len(b) >= self.chunk_size:
            # _textiowrapper_writeflush() calls buffer.write().
            # self->pending_bytes can be appended during buffer->write()
            # or other thread.
            # We need to loop until buffer becomes empty.
            # https://github.com/python/cpython/issues/118138
            # https://github.com/python/cpython/issues/119506
            while self.pending_bytes:
                self._writeflush(space)

        if not self.pending_bytes:
            self.pending_bytes = [b]
        else:
            self.pending_bytes.append(b)
                
        self.pending_bytes_count += len(b)

        if (self.pending_bytes_count >= self.chunk_size or
            needflush or text_needflush):
            self._writeflush(space)

        if needflush:
            space.call_method(self.w_buffer, "flush")

        if self.snapshot:
            # CPython GH-35928
            self.decoded.reset()
            self.snapshot = None

        if self.w_decoder:
            space.call_method(self.w_decoder, "reset")

        return space.newint(textlen)

    def _writeflush(self, space):
        # jit inlinable fast path
        if not self.pending_bytes:
            return

        self._really_flush(space)

    def _really_flush(self, space):
        pending_bytes = ''.join(self.pending_bytes)
        self.pending_bytes = None
        self.pending_bytes_count = 0

        while True:
            try:
                space.call_method(self.w_buffer, "write",
                                  space.newbytes(pending_bytes))
            except OperationError as e:
                if trap_eintr(space, e):
                    continue
                raise
            else:
                break

    def detach_w(self, space):
        self._check_attached(space)
        space.call_method(self, "flush")
        w_buffer = self.w_buffer
        self.w_buffer = None
        self.state = STATE_DETACHED
        return w_buffer

    # _____________________________________________________________
    # seek/tell

    def _decoder_setstate(self, space, cookie):
        # When seeking to the start of the stream, we call decoder.reset()
        # rather than decoder.getstate().
        # This is for a few decoders such as utf-16 for which the state value
        # at start is not (b"", 0) but e.g. (b"", 2) (meaning, in the case of
        # utf-16, that we are expecting a BOM).
        if cookie.start_pos == 0 and cookie.dec_flags == 0:
            space.call_method(self.w_decoder, "reset")
        else:
            space.call_method(self.w_decoder, "setstate",
                              space.newtuple2(space.newbytes(""),
                                              space.newint(cookie.dec_flags)))

    def _encoder_reset(self, space, start_of_stream):
        if start_of_stream:
            space.call_method(self.w_encoder, "reset")
            self.encoding_start_of_stream = True
        else:
            space.call_method(self.w_encoder, "setstate", space.newint(0))
            self.encoding_start_of_stream = False

    def _encoder_setstate(self, space, cookie):
        self._encoder_reset(space,
                            cookie.start_pos == 0 and cookie.dec_flags == 0)

    @unwrap_spec(whence=int)
    def seek_w(self, space, w_pos, whence=0):
        self._check_attached(space)

        if not self.seekable:
            self._unsupportedoperation(space,
                                       "underlying stream is not seekable")

        if whence == 1:
            # seek relative to current position
            if not space.eq_w(w_pos, space.newint(0)):
                self._unsupportedoperation(
                    space, "can't do nonzero cur-relative seeks")
            # Seeking to the current position should attempt to sync the
            # underlying buffer with the current position.
            w_pos = space.call_method(self, "tell")

        elif whence == 2:
            # seek relative to end of file
            if not space.eq_w(w_pos, space.newint(0)):
                self._unsupportedoperation(
                    space, "can't do nonzero end-relative seeks")
            space.call_method(self, "flush")
            self.decoded.reset()
            self.snapshot = None
            if self.w_decoder:
                space.call_method(self.w_decoder, "reset")
            w_res = space.call_method(self.w_buffer, "seek",
                                      w_pos, space.newint(whence))
            if self.w_encoder:
                # If seek() == 0, we are at the start of stream
                start_of_stream = space.eq_w(w_res, space.newint(0))
                self._encoder_reset(space, start_of_stream)
            return w_res

        elif whence != 0:
            raise oefmt(space.w_ValueError,
                        "invalid whence (%d, should be 0, 1 or 2)",
                        whence)

        if space.is_true(space.lt(w_pos, space.newint(0))):
            raise oefmt(space.w_ValueError,
                        "negative seek position %R", w_pos)

        space.call_method(self, "flush")

        # The strategy of seek() is to go back to the safe start point and
        # replay the effect of read(chars_to_skip) from there.
        cookie = PositionCookie(space.bigint_w(w_pos))

        # Seek back to the safe start point
        space.call_method(self.w_buffer, "seek", space.newint(cookie.start_pos))

        self.decoded.reset()
        self.snapshot = None

        # Restore the decoder to its state from the safe start point.
        if self.w_decoder:
            self._decoder_setstate(space, cookie)

        if cookie.chars_to_skip:
            # Just like _read_chunk, feed the decoder and save a snapshot.
            w_chunk = space.call_method(self.w_buffer, "read",
                                        space.newint(cookie.bytes_to_feed))
            if not space.isinstance_w(w_chunk, space.w_bytes):
                msg = "underlying read() should have returned " \
                      "a bytes object, not '%T'"
                raise oefmt(space.w_TypeError, msg, w_chunk)

            self.snapshot = PositionSnapshot(cookie.dec_flags,
                                             space.bytes_w(w_chunk))

            w_decoded = space.call_method(self.w_decoder, "decode",
                                          w_chunk, space.newbool(bool(cookie.need_eof)))
            w_decoded = check_decoded(space, w_decoded)

            # Skip chars_to_skip of the decoded characters
            if space.len_w(w_decoded) < cookie.chars_to_skip:
                raise oefmt(space.w_IOError,
                            "can't restore logical file position")
            self.decoded.set(space, w_decoded)
            self.decoded.pos = w_decoded._index_to_byte(cookie.chars_to_skip)
            self.decoded.upos = cookie.chars_to_skip
        else:
            self.snapshot = PositionSnapshot(cookie.dec_flags, "")

        # Finally, reset the encoder (merely useful for proper BOM handling)
        if self.w_encoder:
            self._encoder_setstate(space, cookie)

        return w_pos

    def tell_w(self, space):
        self._check_closed(space)
        if not self.seekable:
            self._unsupportedoperation(space,
                                       "underlying stream is not seekable")
        if not self.telling:
            raise oefmt(space.w_IOError,
                        "telling position disabled by next() call")

        self._writeflush(space)
        space.call_method(self, "flush")

        w_pos = space.call_method(self.w_buffer, "tell")

        if self.w_decoder is None or self.snapshot is None:
            assert not self.decoded.text
            return w_pos

        cookie = PositionCookie(space.bigint_w(w_pos))

        # Skip backward to the snapshot point (see _read_chunk)
        cookie.dec_flags = self.snapshot.flags
        input = self.snapshot.input
        if len(input) > cookie.start_pos:
            # Must be a device or something that reports 0 for tell()
            # CPython GH-95782
            return space.newint(0)
        cookie.start_pos -= len(input)

        # How many decoded characters have been used up since the snapshot?
        if self.decoded.pos == 0:
            # We haven't moved from the snapshot point.
            return space.newlong_from_rbigint(cookie.pack())

        # chars_to_skip = codepoints_in_utf8(
        #   self.decoded.text, end=self.decoded.pos)
        chars_to_skip = self.decoded.upos

        # Starting from the snapshot position, we will walk the decoder
        # forward until it gives us enough decoded characters.
        w_saved_state = space.call_method(self.w_decoder, "getstate")

        skip_bytes = 0
        # Fast search for an acceptable start point, close to our
        # current pos
        skip_bytes = int(self.b2cratio * chars_to_skip)
        skip_back = 1;
        # ??? this assert is in CPython but is not triggered??
        # assert skip_back <= len(input)
        while skip_bytes > 0:
            # Decode up to temptative start point
            self._decoder_setstate(space, cookie)

            w_decoded = space.call_method(self.w_decoder, "decode",
                                          space.newbytes(input[0:skip_bytes])
                                          )
            check_decoded(space, w_decoded)
            chars_decoded = space.len_w(w_decoded)
            if chars_decoded <= chars_to_skip:
                w_state = space.call_method(self.w_decoder, "getstate")
                w_dec_buffer, w_flags = space.unpackiterable(w_state, 2)
                dec_buffer_len = space.len_w(w_dec_buffer)
                if (dec_buffer_len == 0):
                    # Before pos and no bytes buffered in decoder => OK
                    cookie.dec_flags = space.int_w(w_flags)
                    chars_to_skip -= chars_decoded
                    break
                # Skip back by buffered amount and reset heuristic
                skip_bytes -= dec_buffer_len;
                skip_back = 1;
            else:
                # We're too far ahead, skip back a bit
                skip_bytes -= skip_back;
                skip_back *= 2;
        if skip_bytes <= 0:
            skip_bytes = 0
            self._decoder_setstate(space, cookie)
        cookie.start_pos += skip_bytes
        cookie.chars_to_skip = chars_to_skip
        if chars_to_skip == 0:
            space.call_method(self.w_decoder, "setstate", w_saved_state)
            return space.newlong_from_rbigint(cookie.pack())
        try:
            # Note our initial start point
            self._decoder_setstate(space, cookie)

            # Feed the decoder one byte at a time.  As we go, note the nearest
            # "safe start point" before the current location (a point where
            # the decoder has nothing buffered, so seek() can safely start
            # from there and advance to this location).

            chars_decoded = 0
            i = skip_bytes
            while i < len(input):
                w_decoded = space.call_method(self.w_decoder, "decode",
                                              space.newbytes(input[i]))
                check_decoded(space, w_decoded)
                chars_decoded += space.len_w(w_decoded)

                cookie.bytes_to_feed += 1

                w_state = space.call_method(self.w_decoder, "getstate")
                w_dec_buffer, w_flags = space.unpackiterable(w_state, 2)
                dec_buffer_len = space.len_w(w_dec_buffer)

                if dec_buffer_len == 0 and chars_decoded <= chars_to_skip:
                    # Decoder buffer is empty, so this is a safe start point.
                    cookie.start_pos += cookie.bytes_to_feed
                    chars_to_skip -= chars_decoded
                    assert chars_to_skip >= 0
                    cookie.dec_flags = space.int_w(w_flags)
                    cookie.bytes_to_feed = 0
                    chars_decoded = 0
                if chars_decoded >= chars_to_skip:
                    break
                i += 1
            else:
                # We didn't get enough decoded data; signal EOF to get more.
                w_decoded = space.call_method(self.w_decoder, "decode",
                                              space.newbytes(""),
                                              space.newint(1))  # final=1
                check_decoded(space, w_decoded)
                chars_decoded += space.len_w(w_decoded)
                cookie.need_eof = 1

                if chars_decoded < chars_to_skip:
                    raise oefmt(space.w_IOError,
                        "can't reconstruct logical file position")
        finally:
            space.call_method(self.w_decoder, "setstate", w_saved_state)

        # The returned cookie corresponds to the last safe start point.
        cookie.chars_to_skip = chars_to_skip
        return space.newlong_from_rbigint(cookie.pack())

    def chunk_size_get_w(self, space):
        self._check_attached(space)
        return space.newint(self.chunk_size)

    def chunk_size_set_w(self, space, w_size):
        self._check_attached(space)
        size = space.int_w(w_size)
        if size <= 0:
            raise oefmt(space.w_ValueError,
                        "a strictly positive integer is required")
        self.chunk_size = size

W_TextIOWrapper.typedef = TypeDef(
    '_io.TextIOWrapper', W_TextIOBase.typedef,
    __new__ = generic_new_descr(W_TextIOWrapper),
    __init__  = interp2app(W_TextIOWrapper.descr_init),
    __repr__ = interp2app(W_TextIOWrapper.descr_repr),
    __next__ = interp2app(W_TextIOWrapper.next_w),
    __getstate__ = interp2app(W_TextIOWrapper.getstate_w),

    read = interp2app(W_TextIOWrapper.read_w),
    readline = interp2app(W_TextIOWrapper.readline_w),
    write = interp2app(W_TextIOWrapper.write_w),
    seek = interp2app(W_TextIOWrapper.seek_w),
    tell = interp2app(W_TextIOWrapper.tell_w),
    detach = interp2app(W_TextIOWrapper.detach_w),
    flush = interp2app(W_TextIOWrapper.flush_w),
    truncate = interp2app(W_TextIOWrapper.truncate_w),
    close = interp2app(W_TextIOWrapper.close_w),

    reconfigure = interp2app(W_TextIOWrapper.reconfigure),

    line_buffering = interp_attrproperty("line_buffering", W_TextIOWrapper,
        wrapfn="newbool"),
    write_through = interp_attrproperty("write_through", W_TextIOWrapper,
        wrapfn="newbool"),
    readable = interp2app(W_TextIOWrapper.readable_w),
    writable = interp2app(W_TextIOWrapper.writable_w),
    seekable = interp2app(W_TextIOWrapper.seekable_w),
    isatty = interp2app(W_TextIOWrapper.isatty_w),
    fileno = interp2app(W_TextIOWrapper.fileno_w),
    _dealloc_warn = interp2app(W_TextIOWrapper._dealloc_warn_w),
    name = GetSetProperty(W_TextIOWrapper.name_get_w),
    buffer = interp_attrproperty_w("w_buffer", cls=W_TextIOWrapper),
    closed = GetSetProperty(W_TextIOWrapper.closed_get_w),
    errors = interp_attrproperty_w("w_errors", cls=W_TextIOWrapper),
    newlines = GetSetProperty(W_TextIOWrapper.newlines_get_w),
    _CHUNK_SIZE = GetSetProperty(
        W_TextIOWrapper.chunk_size_get_w, W_TextIOWrapper.chunk_size_set_w
    ),
)