Apple-2-SW/cppo-ng
1
0
Fork 0
mirror of https://github.com/iKarith/cppo-ng.git synced 2025-02-12 23:32:03 +00:00
Code Issues Projects Releases Wiki Activity

Make buffer more stylistically consistent

Browse Source 
We're beginning to develop some consistent styling in blocksfree/cppo-ng, but
the buffer code predates most of that.  Let's fix that.
This commit is contained in:
T. Joseph Carter 2017-07-20 04:19:09 -07:00
parent d3c3e1479c
commit 7c169b2053
2 changed files with 144 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

127
blocksfree/buffer/buffertype.py
View File

@ -16,32 +16,34 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Abstract base class for disk image buffers
The buffers we will work with most commonly represent 140k disk images. For
that, it'd make a lot of sense to simply use a bytearray. It's built in to
the language and pretty efficient for an interpreted OO construct. But even
ProDOS disk images can be 32M in size.
Well that's fine, because most systems have 1G or more of RAM, plus swap, you
might say. And that's true, they do have that much RAM. Only on an embedded
system like the Raspberry Pi, they may not have much or any swap. And while
ProDOS volumes will never be larger than 32M, GS/OS supports other
filesystems
HFS is even commonly used. It's true that HFS support has not been something
you could expect from AppleCommander, to say nothing of cppo, but we ought to
consider the possibility for the future. As of this writing, we cannot
really guarantee that your host operating system can handle HFS completely.
How do you access resource fork on Linux? How do you access any of it on
Windows? Can Apple OSes officially carrying the designation "macOS" even
open old HFS volumes read-only anymore? They haven't had read-write access
for some time now.
"""
from abc import ABCMeta, abstractmethod
from typing import Optional
# Rationale:
#
# The buffers we will work with most commonly represent 140k disk images. For
# that, it'd make a lot of sense to simply use a bytearray. It's built in to
# the language and pretty efficient for an interpreted OO construct. But even
# ProDOS disk images can be 32M in size.
#
# Well that's fine, because most systems have 1G or more of RAM, plus swap, you
# might say. And that's true, they do have that much RAM. Only on an embedded
# system like the Raspberry Pi, they may not have much or any swap. And while
# ProDOS volumes will never be larger than 32M, GS/OS supports other
# filesystems
#
# HFS is even commonly used. It's true that HFS support has not been something
# you could expect from AppleCommander, to say nothing of cppo, but we ought to
# consider the possibility for the future. As of this writing, we cannot
# really guarantee that your host operating system can handle HFS completely.
# How do you access resource fork on Linux? How do you access any of it on
# Windows? Can Apple OSes officially carrying the designation "macOS" even
# open old HFS volumes read-only anymore? They haven't had read-write access
# for some time now.
class BufferType(metaclass=ABCMeta):
class BufferType(object, metaclass=ABCMeta):
"""Abstract class that describes a BufferType.
Read-only BufferType subclasses must implement read and __len__, as well as
@ -61,41 +63,49 @@ class BufferType(metaclass=ABCMeta):
@abstractmethod
def __len__(self) -> int:
"""Return len(self)"""
"""Implement len(self)
Subclasses must provide an implementation for this method.
"""
pass
@property
def changed(self):
"""Returns True if buffer has been altered
"""Return True if buffer has been altered
Always returns false for read-only buffers
Returns:
Always False for read-only buffers
"""
return False
@abstractmethod
def read(
self,
start: int = 0,
count: Optional[int] = None,
limit: bool = True
) -> bytearray:
"""Return bytearray of count bytes from buffer beginning at start
def read(self, start: int, count: int) -> bytes:
"""Return count bytes from buffer beginning at start
Should raise IndexError if an attempt to read past the end of the
buffer is made.
Args:
start: Starting position of bytes to return
count: Number of bytes to return
Returns:
bytes object of the requested length copied from buffer
Raises:
IndexError if attempt to read outside the buffer is made
"""
pass
@abstractmethod
def read1(
self,
offset: int = 0,
limit: bool = True
) -> int:
"""Return int of single byte from buffer at offset
def read1(self, offset: int) -> int:
"""Return single byte from buffer as int
Should raise IndexError if an attempt to read past the end of the
buffer is made.
Args:
offset: The position of the requested byte in the buffer
Returns:
int value of the requested byte
Raises:
IndexError if attempt to read outside the buffer is made
"""
pass
@ -103,26 +113,43 @@ class BufferType(metaclass=ABCMeta):
self,
buf: bytes,
start: int,
count: Optional[int] = None,
limit: bool = True
count: Optional[int] = None
) -> None:
"""Writes bytes to buffer beginning at start
"""Write given bytes-like object to buffer at start
Raises NotImplementedError unless implemented by subclass
Subclasses should raise IndexError if an attempt to write outside the
buffer is made.
Args:
buf: The bytes-like object to write
start: Offset to where in buffer it should be written
count: Length to write (default: length of buf)
Raises:
NotImplementedEror unless implemented by subclass
"""
raise NotImplementedError('buffer does not support writing')
def resize(self, size: int) -> None:
"""Resize buffer to size
r"""Resize a given buffer
Raises NotImplementedError unless implemented by subclass
Resizes the current buffer in place. If size < len(self), the buffer
will be truncated. If size > len(self), the buffer will be extended.
The newly added bytes will be b'\x00'
Args:
size: New size of buffer
Raises:
NotImplementedError unless implemented by subclass
"""
raise NotImplementedError('buffer does not support writing')
@property
def locked(self) -> bool:
"""Returns True for read-only buffers.
"""Determine writability of buffer
Returns True unless writing is implemented by subclass
Returns:
True unless writing is implemented by subclass
"""
return True

111
blocksfree/buffer/bytebuffer.py
View File

@ -15,6 +15,8 @@
# 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
"""Read/Write BufferType that lives in memory"""
from typing import Dict, List, Optional, Union
from .buffertype import BufferType
@ -30,7 +32,7 @@ class ByteBuffer(BufferType):
def __init__(
self,
bytes_or_int: Union[bytes,int],
bytes_or_int: Union[bytes, int],
changed: bool = False,
locked: bool = False
) -> None:
@ -39,45 +41,53 @@ class ByteBuffer(BufferType):
self._locked = locked
def __len__(self) -> int:
"""Return len(self)"""
"""Implement len(self)"""
return len(self._buf)
def read(
self,
start: int = 0,
count: int = 1,
limit: bool = True
) -> bytearray:
"""Return bytearray of count bytes from buffer beginning at start
@property
def changed(self):
"""Return True if buffer has been altered
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
Returns:
Always False for read-only buffers
"""
return self._changed
def read(self, start: int, count: int) -> bytes:
"""Return count bytes from buffer beginning at start
Args:
start: Starting position of bytes to return
count: Number of bytes to return
Returns:
bytes object of the requested length copied from buffer
Raises:
IndexError if attempt to read outside the buffer is made
"""
if count is None:
count = len(self) - start
try:
assert(start >= 0)
assert(count >= 0)
if limit == True:
assert(start + count <= len(self._buf))
assert start >= 0
assert count >= 0
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 read1(
self,
offset: int = 0,
limit: bool = True
) -> int:
"""Return int of single byte from buffer at offset
def read1(self, offset: int) -> int:
"""Return single byte from buffer as int
Should raise IndexError if an attempt to read past the end of the
buffer is made.
Args:
offset: The position of the requested byte in the buffer
Returns:
int value of the requested byte
Raises:
IndexError if attempt to read outside the buffer is made
"""
try:
if limit == True:
assert 0 <= offset <= len(self._buf)
assert 0 <= offset <= len(self._buf)
except AssertionError:
raise IndexError('buffer read with index out of range')
return self._buf[offset]
@ -86,14 +96,17 @@ class ByteBuffer(BufferType):
self,
buf: bytes,
start: int,
count: Optional[int] = None,
limit: bool = True
count: Optional[int] = None
) -> None:
"""Writes bytes to buffer beginning at start
"""Write given bytes-like object to buffer 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
Args:
buf: The bytes-like object to write
start: Offset to where in buffer it should be written
count: Length to write (default: length of buf)
Raises:
IndexError if attempt to read outside the buffer is made
"""
if self.locked:
raise BufferError('cannot write to locked buffer')
@ -101,21 +114,27 @@ class ByteBuffer(BufferType):
if not count:
count = len(buf)
try:
assert(start >= 0)
assert(count >= 0)
if limit == True:
assert(start + count <= len(self._buf))
assert start >= 0
assert count >= 0
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
def resize(self, size: int) -> None:
r"""Resize a given buffer
If size is larger than len(self), the buffer is appended with zero
bytes. If it is smaller, the buffer will be truncated.
Resizes the current buffer in place. If size < len(self), the buffer
will be truncated. If size > len(self), the buffer will be extended.
The newly added bytes will be b'\x00'
Args:
size: New size of buffer
Raises:
BufferError if buffer is locked
"""
if self.locked:
raise BufferError('cannot write to locked buffer')
@ -127,8 +146,12 @@ class ByteBuffer(BufferType):
self._changed = True
@property
def locked(self):
"""Returns True for read-only buffers."""
def locked(self) -> bool:
"""Determine writability of buffer
Returns:
True if buffer has been locked to prevent writing
"""
return self._locked
@locked.setter