Source code for httk.core.datastream.textstream_file_view

import io
from collections.abc import Iterator
from typing import Any, NoReturn, Self

from ..views import unwrap
from .textstream_api import TextstreamAPI
from .textstream_backend import TextstreamBackend
from .textstream_like import TextstreamLike
from .textstream_view import TextstreamView




class TextstreamFileView(TextstreamView, io.TextIOBase, TextstreamAPI):
    r"""
    A view presenting an underlying data streaming backend via the full io.TextIOBase API, which is a superset of TextstreamAPI.

    :param obj: Text-stream source to present through the text file API.
    :param \**hints: Backend-selection, encoding, and compression hints.
    """

    _backend: TextstreamBackend
    _readline_buffer: str

    def __new__(cls, obj: TextstreamLike, **hints: Any) -> Self:
        if isinstance(obj, cls):
            return obj
        backend = cls._prepare_backend(obj, hints)
        instance = super().__new__(cls)
        instance._backend = backend
        instance._readline_buffer = ""
        return instance

    def __init__(self, obj: TextstreamLike, **hints: Any) -> None:
        pass



    def unwrap(self) -> Any:
        """Return the raw representation of the wrapped backend.

        :return: The backend's most raw available representation.
        """
        return unwrap(self._backend)


    @property


    def name(self) -> str | None:
        """Return the backend's source name when one exists."""
        return self._backend.name


    @property


    def closed(self) -> bool:
        """Report whether the backend is closed."""
        return self._backend.closed




    def close(self) -> None:
        """Close the underlying backend."""
        self._backend.close()




    def readable(self) -> bool:
        """Report that the view supports reading.

        :return: Always ``True``.
        """
        return True




    def writable(self) -> bool:
        """Report that the view does not support writing.

        :return: Always ``False``.
        """
        return False




    def seekable(self) -> bool:
        """Report whether the backend supports seeking and telling.

        :return: Whether both operations are available.
        """
        return hasattr(self._backend, "seek") and hasattr(self._backend, "tell")




    def flush(self) -> None:
        """Flush the backend when it provides flushing.

        :raises ValueError: If the view is closed.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")
        flush = getattr(self._backend, "flush", None)
        if flush:
            flush()




    def read(self, size: int | None = -1) -> str:
        """Read up to ``size`` characters, or all remaining characters when ``size`` is negative.

        :param size: Maximum number of characters to read; ``None`` also means all remaining characters.
        :return: The text read from the stream.
        :raises ValueError: If the view is closed.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")

        if size is None or size < 0:
            if self._readline_buffer:
                prefix = self._readline_buffer
                self._readline_buffer = ""
                return prefix + self._backend.read()
            return self._backend.read()

        if size == 0:
            return ""

        if self._readline_buffer:
            if len(self._readline_buffer) >= size:
                out = self._readline_buffer[:size]
                self._readline_buffer = self._readline_buffer[size:]
                return out
            prefix = self._readline_buffer
            self._readline_buffer = ""
            return prefix + self._backend.read(size - len(prefix))

        return self._backend.read(size)




    def readline(self, size: int | None = -1) -> str:  # type: ignore[override] ## https://github.com/python/mypy/issues/9643
        """Read one line, optionally limited to ``size`` characters.

        :param size: Maximum number of characters to read; ``None`` means no limit.
        :return: The line read, including its newline when present.
        :raises ValueError: If the view is closed.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")
        if size == 0:
            return ""
        if size is None:
            size = -1

        parts: list[str] = []
        total = 0

        while True:
            if self._readline_buffer:
                chunk = self._readline_buffer
                self._readline_buffer = ""
            else:
                to_read = 8192
                if size >= 0:
                    remaining = size - total
                    if remaining <= 0:
                        break
                    to_read = min(to_read, remaining)
                chunk = self._backend.read(to_read)

            if chunk == "":
                break

            newline_pos = chunk.find("\n")
            if newline_pos != -1:
                newline_pos += 1
                take = chunk[:newline_pos]
                rest = chunk[newline_pos:]

                if size >= 0 and total + len(take) > size:
                    cutoff = size - total
                    parts.append(take[:cutoff])
                    self._readline_buffer = take[cutoff:] + rest
                    break

                parts.append(take)
                self._readline_buffer = rest
                break

            if size >= 0 and total + len(chunk) > size:
                cutoff = size - total
                parts.append(chunk[:cutoff])
                self._readline_buffer = chunk[cutoff:]
                break

            parts.append(chunk)
            total += len(chunk)

        return "".join(parts)




    def readlines(self, hint: int = -1) -> list[str]:  # type: ignore[override] ## https://github.com/python/mypy/issues/9643
        """Read lines until EOF or until the accumulated size reaches ``hint``.

        :param hint: Approximate minimum number of characters to collect, or a negative value for no limit.
        :return: The lines read from the stream.
        :raises ValueError: If the view is closed.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")

        lines: list[str] = []
        total = 0

        while True:
            line = self.readline()
            if line == "":
                break
            lines.append(line)
            total += len(line)
            if hint >= 0 and total >= hint:
                break

        return lines


    def __iter__(self) -> Iterator[str]:  # type: ignore[override] ## https://github.com/python/mypy/issues/9643
        """Iterate over the stream one line at a time.

        :return: This stream as its line iterator.
        """
        return self

    def __next__(self) -> str:  # type: ignore[override] ## https://github.com/python/mypy/issues/9643
        """Return the next line from the stream.

        :return: The next line.
        :raises StopIteration: When the stream is exhausted.
        """
        line = self.readline()
        if line == "":
            raise StopIteration
        return line



    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
        """Move the stream position and discard buffered line data.

        :param offset: Position adjustment interpreted according to ``whence``.
        :param whence: Reference point for ``offset``.
        :return: The resulting absolute stream position.
        :raises ValueError: If the view is closed.
        :raises io.UnsupportedOperation: If the backend is not seekable.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")
        seek = getattr(self._backend, "seek", None)
        if not seek:
            raise io.UnsupportedOperation("underlying stream is not seekable")

        pos = seek(offset, whence)
        self._readline_buffer = ""
        return pos




    def tell(self) -> int:
        """Return the logical stream position before buffered line data.

        :return: The logical absolute stream position.
        :raises ValueError: If the view is closed.
        :raises io.UnsupportedOperation: If the backend does not support telling.
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")
        tell = getattr(self._backend, "tell")  # noqa: B009  # tell is an optional dynamic backend capability
        if not tell:
            raise io.UnsupportedOperation("underlying stream does not support tell()")

        pos = tell()
        return pos - len(self._readline_buffer)




    def detach(self) -> NoReturn:
        """Reject detaching because the view owns its backend interface.

        :raises io.UnsupportedOperation: Always, because detaching is unsupported.
        """
        raise io.UnsupportedOperation("detach")


    @property


    def encoding(self) -> str | None:  # type: ignore[override] ## these should not be writable
        """Return the backend's text encoding when available."""
        return getattr(self._backend, "encoding", None)


    @property


    def errors(self) -> str | None:  # type: ignore[override] ## these should not be writable
        """Return the backend's error handling mode when available."""
        return getattr(self._backend, "errors", None)


    @property


    def newlines(self) -> str | tuple[str, ...] | None:  # type: ignore[override] ## these should not be writable
        """Return the newline conventions observed by the backend when available."""
        return getattr(self._backend, "newlines", None)