Metadata-Version: 2.4
Name: cs-buffer
Version: 20260915.1
Summary: Facilities to do with buffers, particularly CornuCopyBuffer, an automatically refilling buffer to support parsing of data streams.
Keywords: python3
Author-email: Cameron Simpson <cs@cskk.id.au>
Description-Content-Type: text/markdown
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)
Requires-Dist: cs.deco>=20260912
Requires-Dist: cs.gimmicks>=20260311
Project-URL: MonoRepo Commits, https://bitbucket.org/cameron_simpson/css/commits/branch/main
Project-URL: Monorepo Git Mirror, https://github.com/cameron-simpson/css
Project-URL: Monorepo Hg/Mercurial Mirror, https://hg.sr.ht/~cameron-simpson/css
Project-URL: Source, https://github.com/cameron-simpson/css/blob/main/lib/python/cs/buffer.py

Facilities to do with buffers, particularly CornuCopyBuffer,
an automatically refilling buffer to support parsing of data streams.



Short summary:


* `CopyingIterator`: Wrapper for an iterator that copies every item retrieved to a callable.


* `CornuCopyBuffer`: An automatically refilling buffer intended to support parsing of data streams.


* `FDIterator`: An iterator over the data of a file descriptor.


* `FileIterator`: An iterator over the data of a file object.


* `SeekableFDIterator`: An iterator over the data of a seekable file descriptor.


* `SeekableFileIterator`: An iterator over the data of a seekable file object.


* `SeekableIteratorMixin`: Mixin supplying a logical with a `seek` method.


* `SeekableMMapIterator`: An iterator over the data of a mappable file descriptor.

# Classes

## class CopyingIterator

Wrapper for an iterator that copies every item retrieved to a callable.

### `CopyingIterator.__init__(self, it, copy_to)`

Initialise with the iterator `it` and the callable `copy_to`.

## class CornuCopyBuffer(cs.deco.Promotable, io.BufferedIOBase)

An automatically refilling buffer intended to support parsing
of data streams.

Its primary purpose is to aid binary parsers
which do not themselves need to handle sources specially;
`CornuCopyBuffer`s are trivially made from `bytes`,
iterables of `bytes` and file-like objects.
See `cs.binary` for convenient parsing classes
which work with `CornuCopyBuffer`s.

A `CornuCopyBuffer` is iterable, yielding data in whatever
sizes come from its `input_data` source, preceeded by any
content in the internal buffer.

A `CornuCopyBuffer` also implements `io.BufferedIOBase` and
so supports file methods such as `.read`, `.tell` and `.seek`
supporting drop in use of the buffer in many file contexts.
Note that backward seeks are not supported. `.seek` will take
advantage of the `input_data`'s `.seek` method if it has one,
otherwise it will use consume the `input_data` as required.

It also supprts `.readline()` and `.readlines()` like a text
file, but the lines are `bytes` ending in `b'\n'`.
Note that as mentioned earlier, iteration yields the natural
`bytes` chunks from the underlying iterator, _and does not
yield "lines"_.

Attributes:
* `buf`: the first of any buffered leading chunks
  buffer of unparsed data from the input, available
  for direct inspection by parsers;
  normally however parsers will use `.extend` and `.take`.
* `offset`: the logical offset of the buffer; this excludes
  buffered data and unconsumed input data

*Note*: the initialiser may supply a cleanup function;
although this will be called via the buffer's `.__del__` method
a prudent user of a buffer should call the `.close()` method
when finished with the buffer to ensure prompt cleanup;
the `contextlib.closing` context manager provides an easy way
to do this in common cases.

The primary methods supporting parsing of data streams are
`.extend()` and `take()`.
Calling `.extend(min_size)` arranges that the internal buffer
contains at least `min_size` bytes.
Calling `.take(size)` fetches exactly `size` bytes from the
internal buffer and the input source if necessary and returns
them, adjusting the internal buffer.

len(`CornuCopyBuffer`) returns the length of any buffered data.

bool(`CornuCopyBuffer`) tests whether len() > 0.

Indexing a `CornuCopyBuffer` accesses the buffered data only,
returning an individual byte's value (an `int`).

### `CornuCopyBuffer.__init__(self, input_data, buf=None, offset=0, seekable=None, copy_offsets=None, copy_chunks=None, close=None, progress=None, final_offset=None)`

Prepare the buffer.

Parameters:
* `input_data`: an iterable of data chunks (`bytes`-like instances);
  if your data source is a file see the `.from_file` factory;
  if your data source is a file descriptor see the `.from_fd`
  factory.
* `buf`: if not `None`, the initial state of the parse buffer
* `offset`: logical offset of the start of the buffer, default `0`
* `seekable`: whether `input_data` has a working `.seek` method;
  the default is `None` meaning that it will be attempted on
  the first skip or seek
* `copy_offsets`: if not `None`, a callable for parsers to
  report pertinent offsets via the buffer's `.report_offset`
  method
* `copy_chunks`: if not `None`, every fetched data chunk is
  copied to this callable

The `input_data` is an iterable whose iterator may have
some optional additional properties:
* `seek`: if present, this is a seek method after the fashion
  of `file.seek`; the buffer's `seek`, `skip` and `skipto`
  methods will take advantage of this if available.
* `offset`: the current byte offset of the iterator; this
  is used during the buffer initialisation to compute
  `input_data_displacement`, the difference between the
  buffer's logical offset and the input data iterable's logical offset;
  if unavailable during initialisation this is presumed to
  be `0`.
* `end_offset`: the end offset of the iterator if known.
* `close`: an optional callable
  that may be provided for resource cleanup
  when the user of the buffer calls its `.close()` method.
* `progress`: an optional `cs.Progress.progress` instance
  to which to report data consumed from `input_data`;
  any object supporting `+=` is acceptable
* `final_offset`: optional `int` specifying the largest
  offset expected to be reached, intended for uses such as
  callers presenting a progress indication; this is, for
  example, provided by `CornuCopyBuffer.from_fd` for regular
  files using the `stat.st_size` field

### `CornuCopyBuffer.__del__(self)`

Release resources when the object is deleted.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.__getitem__(self, index)`

Fetch from the internal buffer.
This does not consume data from the internal buffer.
Note that this is an expensive way to access the buffer,
particularly if `index` is a slice.

If `index` is a `slice`, slice the join of the internal subbuffers.
This is quite expensive
and it is probably better to `take` or `takev`
some data from the buffer.

Otherwise `index` should be an `int` and the corresponding
buffered byte is returned.

This is usually not a very useful method;
its primary use case is to probe the buffer to make a parsing decision
instead of taking a byte off and (possibly) pushing it back.

### `CornuCopyBuffer.__len__(self)`

The length is the length of the internal buffer: data available without a fetch.

### `CornuCopyBuffer.__next__(self)`

Fetch a data chunk from the buffer.

### `CornuCopyBuffer.as_fd(self, maxlength=Ellipsis)`

Create a pipe and dispatch a `Thread` to copy
up to `maxlength` bytes from `bfr` into it.
Return the file descriptor of the read end of the pipe.

The default `maxlength` is `Ellipsis`, meaning to copy all data.

Note that the thread preemptively consumes from the buffer.

This is useful for passing buffer data to subprocesses.

### `CornuCopyBuffer.at_eof(self)`

Test whether the buffer is at end of input.

*Warning*: this will fetch from the `input_data` if the buffer
is empty and so it may block.

### `CornuCopyBuffer.bounded(self, end_offset) -> 'CornuCopyBuffer'`

Return a new `CornuCopyBuffer` operating on a bounded view
of this buffer.

This supports parsing of the buffer contents without risk
of consuming past a certain point, such as the known end
of a packet structure.

Parameters:
* `end_offset`: the ending offset of the new buffer.
  Note that this is an absolute offset, not a length.

The new buffer starts with the same offset as `self` and
use of the new buffer affects `self`. After a flush both
buffers will again have the same offset and the data consumed
via the new buffer will also have been consumed from `self`.

Here is an example.
* Make a buffer `bfr` with 9 bytes of data in 3 chunks.
* Consume 2 bytes, advancing the offset to 2.
* Make a new bounded buffer `subbfr` extending to offset
  5. Its inital offset is also 2.
* Iterate over it, yielding the remaining single byte chunk
  from ``b'abc'`` and then the first 2 bytes of ``b'def'``.
  The new buffer's offset is now 5.
* Try to take 2 more bytes from the new buffer - this fails.
* Flush the new buffer, synchronising with the original.
  The original's offset is now also 5.
* Take 2 bytes from the original buffer, which succeeds.

Example:

    >>> bfr = CornuCopyBuffer([b'abc', b'def', b'ghi'])
    >>> bfr.offset
    0
    >>> bfr.take(2)
    b'ab'
    >>> bfr.offset
    2
    >>> subbfr = bfr.bounded(5)
    >>> subbfr.offset
    2
    >>> for bs in subbfr:
    ...   print(bs)
    ...
    b'c'
    b'de'
    >>> subbfr.offset
    5
    >>> subbfr.take(2)
    Traceback (most recent call last):
        ...
    EOFError: insufficient input data, wanted 2 bytes but only found 0
    >>> subbfr.flush()
    >>> bfr.offset
    5
    >>> bfr.take(2)
    b'fg'

*WARNING*: if the bounded buffer is not completely consumed
then it is critical to call the new `CornuCopyBuffer`'s `.flush`
method to push any unconsumed buffer back into this buffer.
Recommended practice is to always call `.flush` when finished
with the new buffer.
The `CornuCopyBuffer.subbuffer` method returns a context manager
which does this automatically.

Also, because the new buffer may buffer some of the unconsumed
data from this buffer, use of the original buffer should
be suspended.

### `CornuCopyBuffer.buf`

    <property object at 0x1019e37e0>

### `CornuCopyBuffer.byte0(self)`

Consume the leading byte and return it as an `int` (`0`..`255`).

### `CornuCopyBuffer.close(self)`

Close the buffer.
This discards the internal buffer of "read but not consumed" data
and calls the `close` callable supplied when the buffer was
initialised, if any.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.detach(self)`

Supports `io.BufferedIOBase`; raises `io.UnsupportedOperation`.

### `CornuCopyBuffer.end_offset`

    <property object at 0x101afcc20>

### `CornuCopyBuffer.extend(self, min_size, short_ok=False)`

Extend the buffer to at least `min_size` bytes.

If `min_size` is `Ellipsis`, extend the buffer to consume all the input.
This should really only be used with bounded buffers
in order to avoid unconstrained memory consumption.

If there are insufficient data available then an `EOFError`
will be raised unless `short_ok` is true (default `False`)
in which case the updated buffer will be short.

### `CornuCopyBuffer.fileno(self)`

Return the underlying file descriptor (an integer) of the stream if it exists,
otherwise raises `io.UnsupportedOperation`.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.flush(self)`

Flush is a no-op.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.from_bytes(bs, offset=0, length=None, **kw)`

Return a `CornuCopyBuffer` fed from the supplied bytes `bs`
starting at `offset` and ending after `length`.

This is handy for callers parsing using buffers but handed bytes.

Parameters:
* `bs`: the bytes
* `offset`: a starting position for the data; the input
  data will start this far into the bytes
* `length`: the maximium number of bytes to use; the input
  data will be cropped this far past the starting point;
  default: the number of bytes in `bs` after `offset`
Other keyword arguments are passed to the buffer constructor.

### `CornuCopyBuffer.from_cli_filespec(filespec: str, **kw)`

Return a `CornuCopyBuffer` fed from the supplied command
line file specification `filespec`.

If `filespec` is `"-"` return a buffer using `sys.stdin`,
otherwise treat it as a filename.

Note: the use of `sys.stdin` relies on `sys.stdin.fileno()`
because we need to do binary reads and `sys.stdin` is
normally in text mode.

### `CornuCopyBuffer.from_fd(fd, readsize=None, offset=None, final_offset=None, **kw)`

Return a new `CornuCopyBuffer` attached to an open file descriptor.

Internally this constructs a `SeekableFDIterator` for regular
files or an `FDIterator` for other files, which provides the
iteration that `CornuCopyBuffer` consumes, but also seek
support if the underlying file descriptor is seekable.

Parameters:
* `fd`: the operating system file descriptor
* `readsize`: an optional preferred read size
* `offset`: a starting position for the data; the file
  descriptor will seek to this offset, and the buffer will
  start with this offset
Other keyword arguments are passed to the buffer constructor.

### `CornuCopyBuffer.from_file(f, readsize=None, offset=None, final_offset=None, **kw)`

Return a new `CornuCopyBuffer` attached to an open file.

Internally this constructs a `SeekableFileIterator`, which
provides the iteration that `CornuCopyBuffer` consumes
and also seek support if the underlying file is seekable.

Parameters:
* `f`: the file like object
* `readsize`: an optional preferred read size
* `offset`: a starting position for the data; the file
  will seek to this offset, and the buffer will start with this
  offset
Other keyword arguments are passed to the buffer constructor.

### `CornuCopyBuffer.from_filename(filename: str, offset=None, final_offset=None, **kw)`

Open the file named `filename` and return a new `CornuCopyBuffer`.

If `offset` is provided, skip to that position in the file.
A negative offset skips to a position that far from the end of the file
as determined by its `Stat.st_size`.

Other keyword arguments are passed to the buffer constructor.

### `CornuCopyBuffer.from_mmap(fd, readsize=None, offset=None, **kw)`

Return a new `CornuCopyBuffer` attached to an mmap of an open
file descriptor.

Internally this constructs a `SeekableMMapIterator`, which
provides the iteration that `CornuCopyBuffer` consumes, but
also seek support.

Parameters:
* `fd`: the operating system file descriptor
* `readsize`: an optional preferred read size
* `offset`: a starting position for the data; the file
  descriptor will seek to this offset, and the buffer will
  start with this offset
Other keyword arguments are passed to the buffer constructor.

### `CornuCopyBuffer.hint(self, size)`

Hint that the caller is seeking at least `size` bytes.

If the `input_data` iterator has a `hint` method, this is
passed to it.

### `CornuCopyBuffer.isatty(self)`

Return `True` if underlying file descriptor is a tty;
`False` is there is no underlying file descriptor.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.iter(self, maxlength)`

Yield chunks from the buffer
up to `maxlength` in total
or until EOF if `maxlength` is `Ellipsis`.

### `CornuCopyBuffer.next(self)`

Fetch a data chunk from the buffer.

### `CornuCopyBuffer.peek(self, size, short_ok=False)`

Examine the leading bytes of the buffer without consuming them,
a `take` followed by a `push`.
Returns the bytes.

### `CornuCopyBuffer.promote(obj)`

Promote `obj` to a `CornuCopyBuffer`,
used by the `@cs.deco.promote` decorator.

Promotes:
* `int`: assumed to be a file descriptor of a file open for binary read
* `str`: assumed to be a filesystem pathname
* `bytes` and `bytes`like objects (`Buffer`s): binary data
* has a `.read1` or `.read` method: assume a file open for binary read
* iterable: assumed to be an iterable of `bytes`like objects

### `CornuCopyBuffer.push(self, bs)`

Push the chunk `bs` onto the front of the buffered data.
Rewinds the logical `.offset` by the length of `bs`.

### `CornuCopyBuffer.read(self, size=-1, one_fetch=False)`

Read bytes from the buffer.
Supports `io.BufferedIOBase`.

Parameters:
* `size`: the desired data size
* `one_fetch`: do a single data fetch, default `False`

In `one_fetch` mode the read behaves like a POSIX file read,
returning up to to `size` bytes from a single I/O operation.

### `CornuCopyBuffer.read1(self, size)`

Shorthand method for `self.read(size,one_fetch=True)`.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.readable(self)`

`CornuCopyBuffer`s are readable.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.readall(self)`

Read all the bytes from the buffer.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.readinto(self, b)`

Read from the buffer and write into `b`.
Return the number of bytes read.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.readline(self, size=-1)`

Return a binary "line" from `self`, where a line is defined by
its ending `b'\n'` delimiter.
The final line from a buffer might not have a trailing newline;
`b''` is returned at EOF.

Example:

    >>> bfr = CornuCopyBuffer([b'abc', b'def\nhij'])
    >>> bfr.readline()
    b'abcdef\n'
    >>> bfr.readline()
    b'hij'
    >>> bfr.readline()
    b''
    >>> bfr.readline()
    b''

### `CornuCopyBuffer.readlines(self, hint=-1)`

Read lines from the file.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.report_offset(self, offset)`

Report a pertinent offset.

### `CornuCopyBuffer.seek(self, offset, whence=None, short_ok=False)`

Return the resulting absolute offset.
Supports `io.BufferedIOBase`.

Parameters are as for `io.seek` except as noted below:
* `whence`: (default `os.SEEK_SET`). This method only supports
  `os.SEEK_SET` and `os.SEEK_CUR`, and does not support seeking to a
  lower offset than the current buffer offset.
* `short_ok`: (default `False`). If true, the seek may not reach
  the target if there are insufficent `input_data` - the
  position will be the end of the `input_data`, and the
  `input_data` will have been consumed; the caller must check
  the returned offset to check that it is as expected. If
  false, a `ValueError` will be raised; however, note that the
  `input_data` will still have been consumed.

### `CornuCopyBuffer.seekable(self)`

`CornuCopyBuffer`s are seekable, although not backwards.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.selfcheck(self, msg='')`

Integrity check for the buffer, useful during debugging.

### `CornuCopyBuffer.skip(self, toskip, copy_skip=None, short_ok=False)`

Advance position by `skip_to`. Return the new offset.

Parameters:
* `toskip`: the distance to advance
* `copy_skip`: callable to receive skipped data.
* `short_ok`: default `False`; if true then skip may return before
  `skipto` bytes if there are insufficient `input_data`.

### `CornuCopyBuffer.skipto(self, new_offset, copy_skip=None, short_ok=False)`

Advance to position `new_offset`. Return the new offset.

Parameters:
* `new_offset`: the target offset.
* `copy_skip`: callable to receive skipped data.
* `short_ok`: default `False`; if true then skipto may return before
  `new_offset` if there are insufficient `input_data`.

Return values:
* `buf`: the new state of `buf`
* `offset`: the final offset; this may be short if `short_ok`.

### `CornuCopyBuffer.subbuffer(self, end_offset)`

Context manager wrapper for `.bounded`
which calls the `.flush` method automatically
on exiting the context.

Example:

    # avoid buffer overrun
    with bfr.subbuffer(bfr.offset+128) as subbfr:
        id3v1 = ID3V1Frame.parse(subbfr)
        # ensure the whole buffer was consumed
        assert subbfr.at_eof()

### `CornuCopyBuffer.tail_extend(self, size)`

Extend method for parsers reading "tail"-like chunk streams,
typically raw reads from a growing file.

This may read 0 bytes at EOF, but a future read may read
more bytes if the file grows.
Such an iterator can be obtained from
``cs.fileutils.read_from(..,tail_mode=True)``.

### `CornuCopyBuffer.take(self, size, short_ok=False)`

Return the next `size` bytes.
Other arguments are as for `.extend()`.

This is a thin wrapper for the `.takev` method.

### `CornuCopyBuffer.takev(self, size, short_ok=False) -> List[collections.abc.Buffer]`

Return the next `size` bytes as a list of chunks
(because the internal buffering is also a list of chunks).
Other arguments are as for `.extend()`.

See `.take()` to get a flat chunk instead of a list.

### `CornuCopyBuffer.tell(self)`

Return the current buffer offset.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.write(self, data)`

`CornuCopyBuffer`s are not writable; raises `io.UnsupportedOperation`.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.writeable(self)`

`CornuCopyBuffer`s are not writable.
Supports `io.BufferedIOBase`.

### `CornuCopyBuffer.writelines(self, lines)`

`CornuCopyBuffer`s are not writable; raises `io.UnsupportedOperation`.
Supports `io.BufferedIOBase`.

## class FDIterator(_FetchIterator)

An iterator over the data of a file descriptor.

### `FDIterator.__init__(self, fd: int, offset=None, readsize=None, align=True)`

Initialise the iterator.

Parameters:
* `fd`: file descriptor
* `offset`: the initial logical offset, kept up to date by
  iteration; the default is the current file position.
* `readsize`: a preferred read size; if omitted then
  `DEFAULT_READSIZE` will be stored
* `align`: whether to align reads by default: if true then
  the iterator will do a short read to bring the `offset`
  into alignment with `readsize`; the default is `True`

### `FDIterator.close(self)`

Close `self.fd` if it is nt yet `None`.

## class FileIterator(_FetchIterator, SeekableIteratorMixin)

An iterator over the data of a file object.

### `FileIterator.__init__(self, fp, offset=None, readsize=None, align=False)`

Initialise the iterator.

Parameters:
* `fp`: file object
* `offset`: the initial logical offset, kept up to date by
  iteration; the default is 0.
* `readsize`: a preferred read size; if omitted then
  `DEFAULT_READSIZE` will be stored
* `align`: whether to align reads by default: if true then
  the iterator will do a short read to bring the `offset`
  into alignment with `readsize`; the default is `False`

### `FileIterator.close(self)`

Detach from the file. Does *not* call `fp.close()`.

## class SeekableFDIterator(FDIterator, SeekableIteratorMixin)

An iterator over the data of a seekable file descriptor.

### `SeekableFDIterator.end_offset`

    <property object at 0x101afd530>

## class SeekableFileIterator(FileIterator)

An iterator over the data of a seekable file object.

*Note*: the iterator closes the file on __del__ or if its
.close method is called.

### `SeekableFileIterator.__init__(self, fp, offset=None, **kw)`

Initialise the iterator.

Parameters:
* `fp`: file object
* `offset`: the initial logical offset, kept up to date by
  iteration; the default is the current file position.
* `readsize`: a preferred read size; if omitted then
  `DEFAULT_READSIZE` will be stored
* `align`: whether to align reads by default: if true then
  the iterator will do a short read to bring the `offset`
  into alignment with `readsize`; the default is `False`

### `SeekableFileIterator.seek(self, new_offset, mode=0)`

Move the logical file pointer.

WARNING: moves the underlying file's pointer.

## class SeekableIteratorMixin

Mixin supplying a logical with a `seek` method.

### `SeekableIteratorMixin.seek(self, new_offset, mode=0)`

Move the logical offset.

## class SeekableMMapIterator(_FetchIterator, SeekableIteratorMixin)

An iterator over the data of a mappable file descriptor.

### `SeekableMMapIterator.__init__(self, fd: int, offset=None, readsize=None, align=True)`

Initialise the iterator.

Parameters:
* `offset`: the initial logical offset, kept up to date by
  iteration; the default is the current file position.
* `readsize`: a preferred read size; if omitted then
  `DEFAULT_READSIZE` will be stored
* `align`: whether to align reads by default: if true then
  the iterator will do a short read to bring the `offset`
  into alignment with `readsize`; the default is `True`

### `SeekableMMapIterator.close(self)`

Close the mmap and detach.

### `SeekableMMapIterator.end_offset`

    <property object at 0x101afd300>
