Implement a ByteBuffer BufferType

I explained in the comments on BufferType why I'm doing this, but the nutshell
version is that I anticipate having bigger files to deal with at some point we
won't want to keep in memory.  Otherwise we could just use bytearrays.

The way this is meant to be used (which admittedly isn't clear--would someone
like to submit a patch that improves the docs to clarify?) is that this is
intended to be used as a context.  In other words, using Python's with
statement.  This isn't all that different for a ByteBuffer, but it would be for
a FileBuffer (which doesn't exist yet and won't for awhile.)

Implementation hint for FileBuffer when I get there: If the file is not
explicitly opened read-only, I intend for read-modify-write to be standard
practice from the start.  That'll mean duplicating a file to a temporary one we
can manipulate safely and then at flush time, copying the changes over the
original.  That way you'd always be able to undo your changes by quitting
without saving.  This seems important as blocksfree is likely to serve a lot of
archival duty and you may only get one shot at trying to save an image from a
damaged floppy.  It would be awful if that image were then destroyed by an
accidental exception somewhere in the middle of other operations.  So let's not
go there.
This commit is contained in:
T. Joseph Carter 2017-07-14 10:22:19 -07:00
parent a385b99b2b
commit d0a8dc2584
1 changed files with 118 additions and 0 deletions

View File

@ -0,0 +1,118 @@
# vim: set tabstop=4 shiftwidth=4 noexpandtab filetype=python:
# Copyright (C) 2017 T. Joseph Carter
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
from typing import Optional, Union
from .buffertype import BufferType
class ByteBuffer(BufferType):
"""ByteBuffer(bytes_or_int[, changed[, locked]]) -> ByteBuffer
Create a BufferType object in memory. If an int is provided, the buffer
will be zero-filled. If it is a bytes-type object, the object will be
copied into the buffer.
"""
def __init__(
self,
bytes_or_int: Union[bytes,int],
changed: bool = False,
locked: bool = False
) -> None:
self._buf = bytearray(bytes_or_int)
self._changed = changed
self._locked = locked
def __len__(self) -> int:
"""Return len(self)"""
return len(self._buf)
def read(
self,
start: int = 0,
count: Optional[int] = None,
limit: bool = True
) -> bytearray:
"""Return bytearray of count bytes from buffer beginning at start
By default, an IndexError will be raised if you read past the end of
the buffer. Pass limit=False if reads outside the buffer should just
return trncated or empty results as with python slicing
"""
if count is None:
count = len(self)
try:
assert(start >= 0)
assert(count >= 0)
if limit == True:
assert(start + count <= len(self._buf))
except AssertionError:
raise IndexError('buffer read with index out of range')
return bytes(self._buf[start:start + count])
def write(
self,
buf: bytes,
start: int,
count: Optional[int] = None,
limit: bool = True
) -> None:
"""Writes bytes to buffer beginning at start
If count is not supplied, the entire bytes-like object will be written
to the buffer. An IndexError will be raised if the write would extend
past the end of the buffer. Pass limit=False
"""
if self.locked:
raise BufferError('cannot write to locked buffer')
if not count:
count = len(buf)
try:
assert(start >= 0)
assert(count >= 0)
if limit == True:
assert(start + count <= len(self._buf))
except AssertionError:
raise IndexError('buffer write with index out of range')
self._buf[start:start+count] = buf
self._changed = True
def resize(self, size):
"""Resize buffer to size
If size is larger than len(self), the buffer is appended with zero
bytes. If it is smaller, the buffer will be truncated.
"""
if self.locked:
raise BufferError('cannot write to locked buffer')
if size <= len(self._buf):
del self._buf[size:]
else:
self._buf.append(bytes(size - len(self._buf)))
self._changed = True
@property
def locked(self):
"""Returns True for read-only buffers."""
return self._locked
@locked.setter
def locked(self, value: bool) -> None:
self._locked = value