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'

'. 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 0x10fae7830>

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 0x10fbf4c70>

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 byteslike objects (Buffers): binary data
• has a .read1 or .read method: assume a file open for binary read
• iterable: assumed to be an iterable of byteslike 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)

CornuCopyBuffers 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)

CornuCopyBuffers 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)

CornuCopyBuffers are not writable; raises io.UnsupportedOperation. Supports io.BufferedIOBase.

CornuCopyBuffer.writeable(self)

CornuCopyBuffers are not writable. Supports io.BufferedIOBase.

CornuCopyBuffer.writelines(self, lines)

CornuCopyBuffers 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 0x10fbf5580>

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 0x10fbf5350>