# -*- coding: utf-8 -*-
"""I/O Objects
===============
.. module:: pcapkit.corekit.io
:mod:`pcapkit.corekit.io` contains seekable I/O object
:class:`~pcapkit.corekit.io.SeekableReader`, which is a customised
implementation to :class:`io.BufferedReader`, and the stream proxies
:class:`~pcapkit.corekit.io.PeekableStream`, which adds ``peek`` to a
seekable stream, and :class:`~pcapkit.corekit.io.NamedStream`, which gives
a stream a ``name``, as well as :func:`~pcapkit.corekit.io.advance`, which moves
a stream past more octets than one read or seek can take, and
:func:`~pcapkit.corekit.io.can_seek`, which tells whether a stream really seeks.
"""
import io
import os
import stat
import sys
import tempfile
import weakref
from typing import TYPE_CHECKING, cast
from pcapkit.utilities import config
from pcapkit.utilities.exceptions import (SeekError, TruncateError, UnsupportedCall,
UnsupportedOperation, stacklevel)
from pcapkit.utilities.warnings import SeekWarning, warn
if TYPE_CHECKING:
from io import BytesIO, RawIOBase
from typing import IO, Any, BinaryIO, Iterable, Optional
from typing_extensions import Buffer
__all__ = ['SeekableReader', 'PeekableStream', 'NamedStream', 'advance', 'can_seek']
def _discard_buffer(file: 'IO[bytes]', path: 'str') -> 'None':
"""Close and remove the temporary file a :class:`SeekableReader` saved its buffer to.
Args:
file: The open buffer file.
path: Its path.
"""
file.close()
try:
os.unlink(path)
except FileNotFoundError: # pragma: no cover
pass
[docs]
class SeekableReader(io.BufferedReader):
"""Seekable buffered reader.
A buffered binary stream providing higher-level access to a readable, non seekable
:class:`~io.RawIOBase` raw binary stream. It inherits :class:`~io.BufferedIOBase`.
When reading data from this object, a larger amount of data may be requested from the
underlying raw stream, and kept in an internal buffer. The buffered data can then be returned
directly on subsequent reads.
The constructor creates a :class:`~io.BufferedReader` for the given readable ``raw`` stream and
``buffer_size``. If ``buffer_size`` is omitted, :data:`~io.DEFAULT_BUFFER_SIZE` is used.
Args:
raw: Underlying raw stream.
buffer_size: Buffer size.
buffer_save: Whether to save buffer to file.
buffer_path: Path to save buffer.
stream_closing: Whether the stream should be closed upon exiting.
.. versionchanged:: 1.5.0
:meth:`truncate` raises
:exc:`~pcapkit.utilities.exceptions.UnsupportedOperation`, as
:meth:`writable` returns :data:`False` (:issue:`645`). A refused
:meth:`seek` leaves the position where it was, and a position the
lookback buffer has dropped raises instead of reading from it
(:issue:`643`, :issue:`644`).
.. versionremoved:: 1.5.0
``writeable()``, a misspelling that never overrode
:meth:`io.IOBase.writable`; use :meth:`writable` (:issue:`645`).
"""
if TYPE_CHECKING:
#: Whether the stream should be closed upon exiting.
_closing: 'bool'
#: Whether the stream is closed.
_closed: 'bool'
#: Underlying raw stream.
_stream: 'IO[bytes]'
#: Current position of the stream.
_tell: 'int'
#: Buffer.
_buffer: 'BytesIO'
#: Buffer view.
_buffer_view: 'memoryview'
#: Buffer size.
_buffer_size: 'int'
#: Buffer start position.
_buffer_set: 'int'
#: Buffer current position.
_buffer_cur: 'int'
#: Octets :meth:`seek` reads past the position to find the end of the stream;
#: the buffer size by default.
_read_ahead_size: 'int'
#: Path to save buffer.
_buffer_path: 'str'
#: File to save buffer.
_buffer_file: 'IO[bytes] | None'
#: Closes and removes the buffer file when it is a temporary one.
_buffer_discard: 'weakref.finalize | None'
@property
def closed(self) -> 'bool':
""":data:`True` if the stream is closed."""
return self._closed
@property
def raw(self) -> 'RawIOBase':
"""The underlying raw stream (a :class:`~io.RawIOBase` instance) that
:class:`~io.BufferedIOBase` deals with. This is not part of the :class:`~io.BufferedIOBase`
API and may not exist on some implementations."""
return cast('RawIOBase', self._stream)
@raw.setter
def raw(self, raw: 'RawIOBase', /) -> 'None':
raise UnsupportedCall("can't set attribute")
def __init__(self, raw: 'IO[bytes]', buffer_size: 'int' = io.DEFAULT_BUFFER_SIZE,
buffer_save: 'bool' = False, buffer_path: 'Optional[str]' = None, *,
stream_closing: 'bool' = True) -> 'None':
super().__init__(cast('RawIOBase', raw), buffer_size)
self._closed = False
self._closing = stream_closing
self._stream = raw
self._buffer = io.BytesIO(bytearray(buffer_size))
self._buffer_view = self._buffer.getbuffer()
self._buffer_size = buffer_size
self._read_ahead_size = buffer_size
self._buffer_discard = None
if buffer_save:
if buffer_path is None:
# NOTE: the buffer is read back by name, which Windows refuses for
# a ``NamedTemporaryFile`` while it is open (#2098). A ``mkstemp``
# file it allows; this removes it on close, or at collection.
fd, self._buffer_path = tempfile.mkstemp()
try:
self._buffer_file = open(fd, 'wb', buffering=0)
except BaseException:
os.close(fd)
os.unlink(self._buffer_path)
raise
self._buffer_discard = weakref.finalize(self, _discard_buffer, self._buffer_file,
self._buffer_path)
else:
self._buffer_file = open(buffer_path, 'wb', buffering=0)
self._buffer_path = buffer_path
else:
self._buffer_file = None
self._buffer_path = ''
self._tell = self._buffer_set = self._buffer_cur = 0
def _write_buffer(self, buf: 'bytes', /) -> 'None':
if self._buffer_file is not None:
self._buffer_file.write(buf)
self._buffer_file.flush()
buf_len = len(buf)
old_ptr = self._buffer_cur
self._buffer_cur += buf_len
if self._buffer_cur > self._buffer_size:
if buf_len >= self._buffer_size:
# NOTE: the last ``_buffer_size`` octets, counted from the front rather
# than as ``buf[-self._buffer_size:]``, which for a buffer of no size at
# all is ``buf[-0:]`` -- the whole of ``buf``, not none of it.
self._buffer_view[:] = buf[buf_len - self._buffer_size:]
else:
self._buffer_view[:-buf_len] = self._buffer_view[old_ptr - (self._buffer_size - buf_len):old_ptr]
self._buffer_view[-buf_len:] = buf
self._buffer_set += self._buffer_cur - self._buffer_size
self._buffer_cur = self._buffer_size
# move the pointer to the end of the original contents
self._buffer.seek(-buf_len, io.SEEK_END)
else:
self._buffer_view[old_ptr:self._buffer_cur] = buf
def _fill(self, size: 'int', minimum: 'Optional[int]' = None, /, *,
keep: 'Optional[int]' = None) -> 'bytes':
"""Read up to ``size`` octets from the stream into the buffer, and return them.
Args:
size: Octets to ask for; negative reads to the end of the stream.
minimum: Octets to keep reading for; ``size`` if omitted.
keep: Position the buffer must still hold afterwards; it grows rather than
slide past it.
Returns:
The octets read, fewer than ``minimum`` only once the stream has ended.
Note:
One read may return fewer octets than asked for at any point -- a pipe hands
over what has been written so far, a raw stream what one system call returned
-- and only an empty one means the stream has ended. Taking the first answer
for all there is would end a capture at a chunk boundary (:issue:`1586`), so this
reads until ``minimum`` octets are in hand. Past that it does not wait:
``read1``, where the stream has it, returns what is already there. A
non-blocking stream's :obj:`None` stops the reading. No one read asks for
more than :data:`~pcapkit.utilities.config.READ_CHUNK` octets, so a bogus
length read from a record allocates only what the stream actually holds.
"""
if size < 0:
buf = self._stream.read(size) or b''
self._write_buffer(buf)
return buf
minimum = size if minimum is None else minimum
read = getattr(self._stream, 'read1', self._stream.read)
# NOTE: ``-1`` sets no limit of its own, but one read still goes no further
# than :data:`sys.maxsize`, whatever the setting, past which it raises
# :exc:`OverflowError` rather than coming back short (#1975).
limit = sys.maxsize if config.READ_CHUNK == -1 else min(config.READ_CHUNK, sys.maxsize)
chunks = [] # type: list[bytes]
count = 0
while True:
chunk = read(min(size - count, limit))
if not chunk:
break
chunks.append(chunk)
count += len(chunk)
if count >= minimum:
break
buf = b''.join(chunks)
if keep is not None:
span = self._buffer_set + self._buffer_cur + len(buf) - max(keep, self._buffer_set)
if span > self._buffer_size:
self._truncate_buffer(span)
self._write_buffer(buf)
return buf
[docs]
def _read_ahead(self) -> 'int':
"""Read the stream to :attr:`_read_ahead_size` octets past the position.
Returns:
How far the stream has been read: that many octets past the position, or the
end of the stream if it ended first. :meth:`seek` takes this as the end.
Note:
The buffer slides forward to make the room, so after this only the last
``_buffer_size`` less ``_read_ahead_size`` octets before the position can still
be sought back to -- none, at the default -- unless ``buffer_save`` keeps them.
It grows instead of sliding past the position itself, so a record longer than
the buffer is held whole, as its reader seeks back to its first octet; a bogus
length grows it only as far as the stream goes. A live stream holds the reader
here until the octets arrive, which is why the engines set
:attr:`_read_ahead_size` to the record they are about to read.
"""
end = self._buffer_set + self._buffer_cur
want = self._tell + self._read_ahead_size - end
if want > 0:
end += len(self._fill(want, keep=self._tell))
return end
def _seek_buffer(self) -> 'int':
"""Point the buffer at the current stream position, and say how much it can serve.
The buffer is a sliding window whose octet 0 sits at absolute ``_buffer_set`` and
whose content occupies ``[0:_buffer_cur]``, so ``_buffer_set + _buffer_cur`` is how
far the underlying stream has been consumed. Every buffered read path needs what
follows from that:
* The buffer's own cursor is a **derived** quantity, ``_tell - _buffer_set``, rather
than another piece of state to be maintained. Nothing keeps it in step: a read
served from the stream writes through :attr:`_buffer_view` and leaves the cursor
untouched, and :meth:`_write_buffer` rewinds it to the start of the appended
octets when the window slides. Deriving it here, at each point of use, is what makes the drift
unrepresentable instead of merely repaired afterwards by the next :meth:`seek`.
* What is available is the run from the position to the end of the content,
``_buffer_set + _buffer_cur - _tell`` -- not ``_buffer_cur``, which is measured
from the window's base and so counts octets that lie *behind* the position, and
not that count less one, which is short of both. Over-asking does not fail: the
buffer is allocated full of NUL padding, so it hands that back as though it were
data, with a plausible length that then suppresses the top-up from the stream.
Returns:
The number of octets the buffer can answer for from the current position.
Raises:
SeekError: If the position lies before the window, whose octets are then gone
for good -- the stream cannot be rewound to re-supply them. :meth:`seek`
refuses that position for the same reason; it is reachable here only through
a :meth:`_truncate_buffer` that moved the window's base past a position already
set.
"""
buf_off = self._tell - self._buffer_set
if buf_off < 0:
raise SeekError(f'cannot read before the beginning of the buffer: '
f'{self._tell} < {self._buffer_set}')
self._buffer.seek(buf_off, io.SEEK_SET)
return self._buffer_cur - buf_off
def _take_buffer(self, size: 'int', /, *, line: 'bool' = False) -> 'bytes':
r"""Read up to ``size`` octets from the buffer's cursor, as a copy, and move past them.
Args:
size: Octets to read, no more than :meth:`_seek_buffer` says the buffer can serve.
line: Stop after the first ``b'\n'``, as :meth:`io.BytesIO.readline` does.
Returns:
The octets read.
Note:
The octets are copied out of :attr:`_buffer_view`, not read with
:meth:`io.BytesIO.read`, ``read1`` or ``readline``. On GraalPy those can hand
back :obj:`bytes` that share the memory of a :class:`io.BytesIO` whose buffer
is exported, as this one's is, so the next :meth:`_write_buffer`, which writes
through the view, would change octets that had already been returned
(:issue:`1963`). A slice of the view, copied, is a copy on every interpreter.
"""
buf_off = self._buffer.tell()
buf = self._buffer_view[buf_off:buf_off + size].tobytes()
if line and (eol := buf.find(b'\n')) >= 0:
buf = buf[:eol + 1]
self._buffer.seek(buf_off + len(buf), io.SEEK_SET)
return buf
[docs]
def close(self) -> 'None':
"""Flush and close this stream. This method has no effect if the file is already closed.
Once the file is closed, any operation on the file (e.g. reading or writing) will raise
a :exc:`ValueError`.
As a convenience, it is allowed to call this method more than once; only the first call,
however, will have an effect.
"""
if self.closed:
return
self.flush()
if self._closing:
self._stream.close()
if self._buffer_discard is not None:
self._buffer_discard()
elif self._buffer_file is not None:
self._buffer_file.close()
self._buffer_view.release()
self._buffer.close()
self._closed = True
[docs]
def fileno(self) -> 'int':
"""Return the underlying file descriptor (an integer) of the stream if it exists.
An :exc:`OSError` is raised if the IO object does not use a file descriptor."""
return self._stream.fileno()
[docs]
def flush(self) -> 'None':
"""Flush the write buffers of the stream if applicable. This does nothing for
read-only and non-blocking streams."""
if self._buffer_file is not None:
self._buffer_file.flush()
# NOTE: a stream its owner has closed already has nothing to flush, and asking
# raises -- out of :meth:`close` too, which a finaliser can only print.
if not getattr(self._stream, 'closed', False):
self._stream.flush()
[docs]
def isatty(self) -> 'bool':
"""Return :data:`True` if the stream is interactive (i.e., connected to a
terminal/tty device)."""
return self._stream.isatty()
[docs]
def readable(self) -> 'bool':
"""Return :data:`True` if the stream can be read from. If :data:`False`,
:meth:`read` will raise :exc:`OSError`."""
return self._stream.readable()
[docs]
def readline(self, size: 'int | None' = -1, /) -> 'bytes':
r"""Read and return one line from the stream. If ``size`` is specified, at most
``size`` bytes will be read.
The line terminator is always ``b'\n'`` for binary files; for text files, the
``newline`` argument to :func:`open` can be used to select the line
terminator(s) recognized.
"""
if size is None:
size = -1
if self._tell >= self._buffer_set + self._buffer_cur:
buf = self._stream.readline(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.readline(size)
elif not size:
# NOTE: a request for no octets is answered without consulting the window,
# which has nothing to say about it. Asking anyway refuses a zero-length
# read from a position :meth:`_truncate_buffer` has left behind the window -- a
# refusal over data that was never wanted.
#
# Truthiness rather than ``size == 0`` on purpose. It is the same test for
# every value this method can be reached with -- ``size`` is an ``int`` by
# the time control arrives, and ``False`` is ``0`` -- but :meth:`peek` does
# not normalise a ``None`` its signature does not permit, and there the two
# spellings diverge: ``size == 0`` would send ``None`` on to the window and
# report a type error as a position error. Keeping it falsy keeps that call
# failing as a :exc:`TypeError`.
buf = b''
else:
buf_rem = self._seek_buffer()
buf = self._take_buffer(buf_rem if size < 0 else min(size, buf_rem), line=True)
# NOTE: an unbounded ``readline`` has to keep going until the line ends, and
# capping it at the buffer's content -- which the line need not end inside --
# is what makes the continuation necessary rather than optional here.
size_rem = -1
if not buf.endswith(b'\n') and (size < 0 or (size_rem := size - len(buf)) > 0):
buf_tmp = self._stream.readline(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
self._tell += len(buf)
return buf
[docs]
def readlines(self, hint: 'int' = -1, /) -> 'list[bytes]':
"""Read and return a list of lines from the stream. ``hint`` can be specified to control
the number of lines read: no more lines will be read if the total size (in
bytes/characters) of all lines so far exceeds ``hint``.
``hint`` values of ``0`` or less, as well as :obj:`None`, are treated as no hint.
Note that it's already possible to iterate on file objects using ``for line in file: ...``
without calling :meth:`file.readlines() <readlines>`.
"""
if hint is None or hint <= 0:
lines = [] # type: list[bytes]
while True:
line = self.readline()
if not line:
break
lines.append(line)
return lines
size = 0
lines = []
while size < hint:
line = self.readline(hint - size)
if not line:
break
lines.append(line)
size += len(line)
return lines
[docs]
def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int':
"""Change the stream position to the given byte ``offset``. ``offset`` is interpreted
relative to the position indicated by ``whence``. The default value for ``whence`` is
:data:`~os.SEEK_SET`. Values for ``whence`` are:
* :data:`~os.SEEK_SET` or ``0`` - start of the stream (the default); ``offset`` should
be zero or positive
* :data:`~os.SEEK_CUR` or ``1`` - current stream position; ``offset`` may be negative
* :data:`~os.SEEK_END` or ``2`` - end of the stream; ``offset`` is usually negative
Return the new absolute position.
Note:
The target is computed, then validated, and only then written to ``_tell``. A
branch that assigned the position before deciding whether to accept it would
leave a refused seek having moved it anyway, with the resync below -- the one
thing that puts the buffer's cursor back in step -- skipped on the way out. A
caller that catches the error and reasonably takes the position to be unchanged
would then read from the rejected offset instead, silently and without a second
error.
The negative check applies to every ``whence``, rather than only
:data:`~os.SEEK_SET` looking at its offset. An absolute position below zero
cannot exist under any of them, and it is not the same failure as a position
that has merely slid out of the window -- which is ordinary, and recoverable
with ``buffer_save=True``. Reporting the first as ``negative seek value`` keeps
the two distinguishable, and, being a property of the position rather than of
the window, it holds with a saved buffer as well, which is exempt from the window
check and would otherwise return a negative position as if the seek had worked.
A non-seekable stream has no end until it is over, so :data:`~os.SEEK_END` is
as far as :meth:`_read_ahead` reads past the position: ``_read_ahead_size``
octets, a buffer's worth by default, or to the real end if that comes first.
The frame readers measure what is left this way, and the engines set that size
to the next record, so a record is measured whole, however long. The buffer's far
edge will not do as the end: once the buffer has filled it is just how far the
stream has been read, so the first record ending there would be taken for the end
of the capture (:issue:`1586`). The read slides the buffer, so a position before
the one :data:`~os.SEEK_END` was sought from may then be refused, as
:meth:`_read_ahead` says.
"""
if whence == io.SEEK_SET:
target = offset
elif whence == io.SEEK_CUR:
target = self._tell + offset
elif whence == io.SEEK_END:
target = self._read_ahead() + offset
else:
raise SeekError(f'invalid whence ({whence}, should be {io.SEEK_SET}, {io.SEEK_CUR} or {io.SEEK_END})')
# NOTE: the buffer's far edge, read after :meth:`_read_ahead` has slid it, is how far
# a seek may go unwarned. It is not the end of the stream.
buf_end = self._buffer_set + self._buffer_size
if target < 0:
raise SeekError(f'negative seek value {target}')
# NOTE: both refusals read the target rather than ``_tell``, which is what lets them
# run before the position is committed. This one is the window's, so a saved buffer
# -- which can still supply the octets from file -- is exempt from it.
if target < self._buffer_set and self._buffer_file is None:
raise SeekError(f'cannot seek before the beginning of the buffer: {target} < {self._buffer_set}')
self._tell = target
if self._tell >= self._buffer_set:
ended = False
if self._tell > (tmp_end := self._buffer_set + self._buffer_cur):
# NOTE: if we do need to seek beyond the existing contents,
# then we'll do a quick read to make up the contents; the
# size of the read is set to be 1/4 size of the buffer or
# the size of the content to be read, whichever is larger.
# However, the length to fill must not be larger than the
# buffer size itself.
#
# The gap is waited for and the rest is not, and the position is
# the target, or the end of the stream if that comes first -- not
# wherever an over-asking read happened to stop (#1586).
tmp_gap = min(self._tell - tmp_end, self._buffer_size)
tmp_len = min(max(tmp_gap, self._buffer_size // 4), self._buffer_size)
tmp_buf = self._fill(tmp_len, tmp_gap)
ended = len(tmp_buf) < tmp_gap
self._tell = min(self._tell, tmp_end + len(tmp_buf))
# NOTE: past a stream that ended inside the buffer, this is the silent seek past
# the end of a file, which a frame reader makes over a record the capture cut
# short. A seek that went past the buffer's edge, or stopped short of a target
# beyond it, is still warned of.
if target > buf_end and not (ended and self._tell <= buf_end):
warn(f'seek beyond the end of the buffer: {target} > {buf_end}',
SeekWarning, stacklevel=stacklevel())
self._buffer.seek(self._tell - self._buffer_set, io.SEEK_SET)
else:
# NOTE: only a saved buffer reaches here -- the refusal above has already
# turned away a position before the window when there is no file to serve it
# from.
self._buffer.seek(0, io.SEEK_SET)
return self._tell
[docs]
def seekable(self) -> 'bool':
"""Return :data:`True` if the stream supports random access. If :data:`False`,
:meth:`seek`, :meth:`tell` and :meth:`truncate` will raise :exc:`OSError`."""
return True
[docs]
def tell(self) -> 'int':
"""Return the current stream position."""
return self._tell
[docs]
def truncate(self, size: 'int | None' = None, /) -> 'int':
"""Resize the stream to the given ``size`` in bytes (or the current position if ``size`` is
not specified).
Raises:
UnsupportedOperation: Always, since the reader is not writable.
Note:
:meth:`io.IOBase.writable` gates this method as well as :meth:`write` -- "If
:data:`False`, :meth:`write` and :meth:`truncate` will raise :exc:`OSError`" -- and
:meth:`writable` here returns :data:`False`, so refusing is what the contract asks
for. It is also what CPython's own read-only buffered readers do: both the
accelerated :class:`io.BufferedReader` and the pure-Python
``_pyio.BufferedReader`` raise :exc:`io.UnsupportedOperation`, the latter from
``_BufferedIOMixin.truncate``'s ``_checkWritable()`` (:issue:`645`).
Resizing the lookback buffer is reachable internally, as :meth:`_truncate_buffer`.
It never touches the underlying stream -- it resizes a private lookback window --
which is why it stays, under a private name.
"""
raise UnsupportedOperation('truncate')
[docs]
def _truncate_buffer(self, size: 'int | None' = None, /) -> 'int':
"""Resize the lookback buffer to the given ``size`` in bytes (or the current position if
``size`` is not specified). The current stream position isn't changed. This resizing can
extend or reduce the current buffer size. In case of extension, the new area is
zero-filled. The new buffer size is returned.
Note:
Nothing here writes to the underlying stream -- :meth:`write` raises -- so what
this resizes is the buffer, not the stream behind it. That is why :meth:`truncate`
refuses (:issue:`645`) while this remains: the operation is a private-window one,
not an :class:`io.IOBase` write. The buffer is a sliding window over a stream that
cannot be seeked: its octet 0 sits at absolute offset ``_buffer_set``, its content
occupies ``[0:_buffer_cur]``, and everything past that is padding never read.
As for which octets survive: an extension appends its zero octets at
the **tail**, the new area being by definition the region past the old end. A
reduction below the content keeps the **most recent** ``size`` octets and advances
``_buffer_set`` past the ones it drops, because what this holds is lookback: the
octets it can still answer for are the ones just read, and the stream is already
beyond them. Padding is never kept in preference to content either way.
Advancing ``_buffer_set`` is what keeps ``_buffer_set + _buffer_cur`` equal to how far
the stream has actually been consumed, which :meth:`seek` relies on to decide whether
it may read ahead to fill a gap. A reduction that shrank the window without moving its
base would leave that sum short of the stream, and the next forward :meth:`seek` would
splice in octets from the wrong absolute offset without complaining.
Octets dropped by a reduction are gone for good, since the stream cannot be rewound to
re-supply them. A position left among them is then before the window, which
:meth:`seek` refuses as it refuses any other.
"""
if size is None:
# NOTE: an unspecified size means the current position, following the
# :meth:`io.IOBase.truncate` convention. The buffer
# is indexed relative to ``_buffer_set``, and the position may sit before it
# once a saved buffer has been rewound, in which case nothing is kept.
size = max(self._tell - self._buffer_set, 0)
if size < 0:
raise TruncateError(f'negative size value {size}')
# NOTE: the position isn't changed by a truncation, but rebuilding the buffer
# resets it, so it is read here and put back below -- moved down by whatever
# the window's base moved up, so that it still denotes the same octet.
buffer_pos = self._buffer.tell()
self._buffer_view.release()
# NOTE: only ``[0:_buffer_cur]`` is content. Slicing the buffer itself would
# keep padding that was never read and count it as though it were data.
temp = self._buffer.getvalue()[:self._buffer_cur]
dropped = max(len(temp) - size, 0)
self._buffer = io.BytesIO(temp[dropped:].ljust(size, b'\x00'))
self._buffer_view = self._buffer.getbuffer()
self._buffer.seek(max(buffer_pos - dropped, 0), io.SEEK_SET)
self._buffer_size = size
# NOTE: ``_buffer_set + _buffer_cur`` is unchanged by construction -- the base
# gains exactly what the content pointer loses -- so the stream's consumption
# point still reads correctly out of the pair.
self._buffer_set += dropped
self._buffer_cur = len(temp) - dropped
return self._buffer_size
[docs]
def writable(self) -> 'bool':
"""Return :obj:`True` if the stream supports writing. If :obj:`False`, :meth:`write` and
:meth:`truncate` will raise :exc:`OSError`."""
return False
def writelines(self, lines: 'Iterable[Buffer]', /) -> 'None':
"""Write a list of lines to the stream. Line separators are not added, so it is usual for
each of the lines provided to have a line separator at the end."""
raise UnsupportedOperation('write')
[docs]
def detach(self) -> 'RawIOBase':
"""Separate the underlying raw stream from the buffer and return it.
After the raw stream has been detached, the buffer is in an unusable state.
Some buffers, like :class:`~io.BytesIO`, do not have the concept of a single raw stream to
return from this method. They raise :exc:`~io.UnsupportedOperation`.
"""
if hasattr(self._stream, 'detach'):
return self._stream.detach()
raise UnsupportedOperation('detach')
[docs]
def read(self, size: 'int | None' = -1, /) -> 'bytes':
"""Read and return ``size`` bytes, or if ``size`` is not given or negative, until EOF or if
the read call would block in non-blocking mode."""
if size is None or size < 0:
size = -1
# NOTE: both reads from the stream go through :meth:`_fill`, which keeps reading
# past a short answer, as :meth:`io.BufferedIOBase.read` is specified to: a short
# result here means the end of the stream, and the frame readers take it so.
if self._tell >= self._buffer_set + self._buffer_cur:
buf = self._fill(size)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.read(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check.
buf = b''
else:
buf_rem = self._seek_buffer()
buf = self._take_buffer(buf_rem if size < 0 else min(size, buf_rem))
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
buf += self._fill(size_rem)
self._tell += len(buf)
return buf
[docs]
def read1(self, size: 'int | None' = -1, /) -> 'bytes':
"""Read and return up to ``size`` bytes with only one call on the raw stream. If at least
one byte is buffered, only buffered bytes are returned. Otherwise, one raw stream read call
is made."""
if size is None:
size = -1
if self._tell >= self._buffer_set + self._buffer_cur:
if hasattr(self._stream, 'read1'):
buf = self._stream.read1(size)
else:
buf = self._stream.read(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.read1(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check.
buf = b''
else:
# NOTE: no continuation from the stream when the buffer answered, since
# :meth:`read1` is specified to return only buffered octets if any are.
buf_rem = self._seek_buffer()
buf = self._take_buffer(buf_rem if size < 0 else min(size, buf_rem))
if not buf: # pragma: no cover
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
if hasattr(self._stream, 'read1'):
buf_tmp = self._stream.read1(size_rem)
else:
buf_tmp = self._stream.read(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
self._tell += len(buf)
return buf
[docs]
def readinto(self, b: 'Buffer', /) -> 'int':
"""Read bytes into a pre-allocated, writable :term:`bytes-like object` ``b`` and return the
number of bytes read. For example, ``b`` might be a :obj:`bytearray`.
Like :meth:`read`, multiple reads may be issued to the underlying raw stream, unless the
latter is interactive.
A :exc:`BlockingIOError` is raised if the underlying raw stream is in non blocking-mode,
and has no data available at the moment.
"""
if TYPE_CHECKING:
b = cast('memoryview', b)
buf = self.read(len(b))
buf_len = len(buf)
b[:buf_len] = buf
return buf_len
[docs]
def readinto1(self, b: 'Buffer', /) -> 'int':
"""Read bytes into a pre-allocated, writable :term:`bytes-like object` ``b``, using at most
one call to the underlying raw stream's :meth:`read` (or :meth:`readinto`) method. Return
the number of bytes read.
A :exc:`BlockingIOError` is raised if the underlying raw stream is in non blocking-mode,
and has no data available at the moment.
"""
if TYPE_CHECKING:
b = cast('memoryview', b)
buf = self.read1(len(b))
buf_len = len(buf)
b[:buf_len] = buf
return buf_len
[docs]
def write(self, b: 'Buffer', /) -> 'int':
"""Write the given :term:`bytes-like object`, ``b``, and return the number of bytes written
(always equal to the length of ``b`` in bytes, since if the write fails an :exc:`OSError`
will be raised). Depending on the actual implementation, these bytes may be readily written
to the underlying stream, or held in a buffer for performance and latency reasons.
When in non-blocking mode, a :exc:`BlockingIOError` is raised if the data needed to be
written to the raw stream but it couldn't accept all the data without blocking.
The caller may release or mutate ``b`` after this method returns, so the implementation
should only access ``b`` during the method call.
"""
raise UnsupportedOperation('write')
[docs]
def peek(self, size: 'int' = 0) -> 'bytes':
"""Return bytes from the stream without advancing the position.
At most one single read on the raw stream is done to satisfy the call.
The number of bytes returned may be less or more than requested.
"""
if self._tell >= self._buffer_set + self._buffer_cur:
if hasattr(self._stream, 'peek'):
buf = self._stream.peek(size)
else:
buf = self._stream.read(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.peek(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check. Only
# the buffered branch is short-circuited: the branch above hands a bare
# ``peek()`` to the raw stream, whose own ``peek(0)`` may legitimately
# return a whole buffer's worth, and that is left as it was.
buf = b''
else:
# NOTE: read through the view rather than through ``self._buffer``, whose
# cursor an ordinary read would advance. A preview must leave the position
# alone, and the buffer's cursor is part of the position: advancing it here
# while leaving ``_tell`` untouched would make the *next* buffered read
# start from the wrong octet, with ``tell()`` reporting the right one.
buf_rem = self._seek_buffer()
buf_off = self._buffer.tell()
buf = bytes(self._buffer_view[
buf_off:buf_off + (buf_rem if size < 0 else min(size, buf_rem))])
if not buf and len(buf) < size: # pragma: no cover
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
if hasattr(self._stream, 'peek'):
buf_tmp = self._stream.peek(size_rem)
else:
buf_tmp = self._stream.read(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
return buf
[docs]
def can_seek(stream: 'IO[bytes]') -> 'bool':
"""Whether ``stream`` can really seek, rather than only saying it can.
On Windows before CPython 3.16, a pipe reports :meth:`~io.IOBase.seekable`
and its ``seek`` silently does nothing (:issue:`2097`), so the answer is
trusted only for a regular file, or for a stream with no descriptor to ask.
Args:
stream: The input stream.
"""
if not stream.seekable():
return False
try:
mode = os.fstat(stream.fileno()).st_mode
except (AttributeError, OSError, ValueError):
return True
return stat.S_ISREG(mode)
#: Octets :func:`advance` reads and discards at a time when
#: :data:`~pcapkit.utilities.config.READ_CHUNK` sets no limit.
_DISCARD_CHUNK = 1 << 20
[docs]
def advance(file: 'IO[bytes]', count: 'int', /) -> 'None':
"""Move ``file`` forward by ``count`` octets, however many one read or seek can take.
Args:
file: Stream to move.
count: Octets to move past, counted from the position.
A frame reader asks for a record no further than one read can go,
:data:`sys.maxsize`, so a record the file really holds past that is read only
as far as the limit; the next record still starts where this one declares it
ends (:issue:`1975`). A seekable stream gets there by relative seeks, none
longer than :data:`sys.maxsize`, and may land past its end, as an absolute seek
would. A :class:`SeekableReader`, or a stream that cannot seek -- including
one that only says it can, as a Windows pipe does (:issue:`2108`) -- reads the
octets and discards them, no more than
:data:`~pcapkit.utilities.config.READ_CHUNK` at a time, and stops at the end
of the stream: a :class:`SeekableReader` seek reads at most a buffer's worth
past what it holds, so a longer one would stop short of the target.
"""
# NOTE: a frame reader built on a caller's stream sees it unwrapped (#2108).
if isinstance(file, SeekableReader) or not can_seek(file):
chunk = _DISCARD_CHUNK if config.READ_CHUNK == -1 else config.READ_CHUNK
chunk = min(chunk, sys.maxsize)
while count > 0:
buf = file.read(min(count, chunk))
if not buf:
break
count -= len(buf)
return
while count > 0:
step = min(count, sys.maxsize)
file.seek(step, io.SEEK_CUR)
count -= step
[docs]
class PeekableStream:
"""Give a seekable binary stream without ``peek`` one.
:class:`~pcapkit.foundation.extraction.Extractor` and the PCAP-NG engine
read ahead with ``peek``, which :class:`io.BufferedReader` has and a plain
:class:`typing.IO` such as :class:`io.BytesIO` does not (:issue:`1506`). The
read-ahead is a read and a seek back, and every other attribute is the
stream's own. It is not an :class:`io.IOBase`, so it has no finaliser that
could close the caller's stream behind their back (:issue:`610`).
Copying and pickling follow the stream (:issue:`1524`). :func:`copy.copy` gives a
second proxy over the *same* stream, so the two share one position.
:func:`copy.deepcopy` and :mod:`pickle` copy the stream too, so they work
exactly when the stream itself can be deep-copied or pickled at that
protocol: an :class:`io.BytesIO` pickles at protocol 2 and above, and an
open file not at all.
Arguments:
stream: seekable binary stream
"""
def __init__(self, stream: 'IO[bytes]') -> 'None':
self._stream = stream
[docs]
def peek(self, size: 'int' = 0) -> 'bytes':
"""Return up to ``size`` octets, at least one if any remain, without consuming them.
Arguments:
size: number of octets wanted
Returns:
The octets at the current position.
"""
pos = self._stream.tell()
try:
return self._stream.read(max(size, 1))
finally:
self._stream.seek(pos, io.SEEK_SET)
def __getattr__(self, name: 'str') -> 'Any':
# NOTE: reached only for a name the proxy lacks, and some names are refused
# rather than forwarded. ``_stream`` is the proxy's own: :mod:`copy` and
# :mod:`pickle` build an instance without :meth:`__init__`, so it is not set
# yet, and forwarding would look it up through here again until the
# recursion limit (#1524). A dunder names a protocol of the proxy's, not of
# the stream's: forwarded, the stream's ``__setstate__`` or ``__deepcopy__``
# would be applied to the proxy.
if name == '_stream' or (name.startswith('__') and name.endswith('__')):
raise AttributeError(f'{type(self).__name__!r} object has no attribute {name!r}')
return getattr(self._stream, name)
[docs]
class NamedStream:
"""Read-only proxy that gives a stream the ``name`` attribute.
``pcapfile.savefile.load_savefile`` dereferences ``input_file.name``
unconditionally, on the way into its trace helper. That is fine for the
:class:`~io.BufferedReader` :class:`~pcapkit.foundation.extraction.Extractor`
normally holds, but not for the :class:`~pcapkit.corekit.io.SeekableReader`
it substitutes when the caller supplied a non-seekable stream -- that class
exposes no ``name``, so the load would fail with :exc:`AttributeError` before
a single byte was read.
Args:
stream: Underlying binary stream.
name: Name to report as :attr:`name`.
"""
def __init__(self, stream: 'BinaryIO', name: 'str') -> 'None':
self._stream = stream
#: Name of the underlying stream.
self.name = name
[docs]
def read(self, size: 'int' = -1) -> 'bytes':
"""Read from the underlying stream.
Args:
size: Number of bytes to read; all remaining bytes if negative.
"""
return self._stream.read(size)