Source code for httk.core.datastream.bytestream_file_view

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

from ..views import unwrap
from .bytestream_api import BytestreamAPI
from .bytestream_backend import BytestreamBackend
from .bytestream_like import BytestreamLike
from .bytestream_view import BytestreamView




class BytestreamFileView(BytestreamView, io.IOBase, BytestreamAPI):
    r"""
    A view presenting an underlying data streaming backend via an io.IOBase-like API.

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

    _backend: BytestreamBackend
    _readline_buffer: bytes

    def __new__(cls, obj: BytestreamLike, **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 = b""
        return instance

    def __init__(self, obj: BytestreamLike, **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) -> bytes:
        """Read up to ``size`` bytes, or all remaining bytes when ``size`` is negative.

        :param size: Maximum number of bytes to read; ``None`` also means all remaining bytes.
        :return: The bytes 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 = b""
                return prefix + self._backend.read()
            return self._backend.read()

        if size == 0:
            return b""

        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 = b""
            return prefix + self._backend.read(size - len(prefix))

        return self._backend.read(size)




    def readline(self, size: int | None = -1) -> bytes:
        """Read one line, optionally limited to ``size`` bytes.

        :param size: Maximum number of bytes 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 b""
        if size is None:
            size = -1

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

        while True:
            if self._readline_buffer:
                chunk = self._readline_buffer
                self._readline_buffer = b""
            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 == b"":
                break

            newline_pos = chunk.find(b"\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 b"".join(parts)




    def readlines(self, hint: int = -1) -> list[bytes]:
        """Read lines until EOF or until the accumulated size reaches ``hint``.

        :param hint: Approximate minimum number of bytes 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[bytes] = []
        total = 0

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

        return lines


    def __iter__(self) -> Iterator[bytes]:
        """Iterate over the stream one line at a time.

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

    def __next__(self) -> bytes:
        """Return the next line from the stream.

        :return: The next line.
        :raises StopIteration: When the stream is exhausted.
        """
        line = self.readline()
        if line == b"":
            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 = b""
        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", None)
        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")