SparseMemoryIO

class hexrec.utils.SparseMemoryIO(memory=None, seek=None)

Sparse memory I/O wrapper.

With respect to the parent class bytesparse.io.MemoryIO, it allows reading and writing memory holes.

Such holes are marked by the following integer values (instead of None):

• 0x100 = hole byte within memory span

  (bytesparse.base.ImmutableMemory.span);

• 0x101 = hole byte before memory start address

  (bytesparse.base.ImmutableMemory.start);

• 0x102 = hole byte after memory end address

  (bytesparse.base.ImmutableMemory.endex);

These special values allow displaying dedicated stuff when dumping memory data to standard output.

See also

bytesparse.io.MemoryIO bytesparse.base.ImmutableMemory.span bytesparse.base.ImmutableMemory.start bytesparse.base.ImmutableMemory.endex

Attributes

closed	Closed stream.
memory	Underlying memory object.

Methods

__init__
close	Closes the stream.
detach	Detaches the underlying raw stream.
fileno	File descriptor identifier.
flush	Flushes buffered data into the underlying raw steam.
getbuffer	Memory view of the underlying memory object.
getvalue	Byte string copy of the underlying memory object.
isatty	Interactive console stream.
peek	Previews the next chunk of bytes.
read	Reads a chunk of bytes.
read1	Reads a chunk of bytes.
readable	Stream is readable.
readinto	Reads data into a byte buffer.
readinto1	Reads data into a byte buffer.
readline	Reads a line.
readlines	Reads a list of lines.
seek	Changes the current stream position.
seekable	Stream is seekable.
skip_data	Skips a data block.
skip_hole	Skips a memory hole.
tell	Current stream position.
truncate	Truncates stream.
writable	Stream is writable.
write	Writes data into the stream.
writelines	Writes lines to the stream.

__del__()

Prepares the object for destruction.

It makes sure the stream is closed upon object destruction.

__enter__()

Context manager enter function.

Returns:

MemoryIO – The stream object itself.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> with MemoryIO(Memory.from_bytes(b'Hello, World!')) as stream:
...     data = stream.read()
>>> data
b'Hello, World!'

__exit__(exc_type, exc_val, exc_tb)

Context manager exit function.

It makes sure the stream is closed upon context exit.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> with MemoryIO(Memory.from_bytes(b'Hello, World!')) as stream:
...     print(stream.closed)
False
>>> print(stream.closed)
True

__init__(memory=None, seek=None)

__iter__()

Iterates over lines.

Repeatedly calls readline(), as long as it returns byte strings. Yields the values returned by such calls.

Yields:

bytes – Single line; terminator included.

See also

readline()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello\nWorld!'], [20, b'Bye\n'], [28, b'Bye!']]
>>> with MemoryIO(Memory.from_blocks(blocks)) as stream:
...     lines = [line for line in stream]
>>> lines
[b'Hello\n', b'World!', b'Bye\n', b'Bye!']

__new__(**kwargs)

__next__()

Next iterated line.

Calls readline() once, returning the value.

Returns:

bytes or int – Line read from the stream, or the negative gap size.

See also

readline()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello\nWorld!'], [20, b'Bye\n'], [28, b'Bye!']]
>>> with MemoryIO(Memory.from_blocks(blocks), seek=9) as stream:
...     print(next(stream))
b'World!'

_check_closed()

Checks if the stream is closed.

In case the stream is closed, it raises ValueError.

Raises:

ValueError – The stream is closed.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> with MemoryIO(Memory.from_bytes(b'ABC')) as stream:
...     stream._check_closed()
>>> stream._check_closed()
Traceback (most recent call last):
    ...
ValueError: I/O operation on closed stream.

close()

Closes the stream.

Any subsequent operations on the closed stream may fail, and some properties may change state.

The stream no more links to an underlying memory object.

See also

closed

Examples

>>> from bytesparse import Memory, MemoryIO

>>> stream = MemoryIO(Memory.from_bytes(b'ABC'))
>>> stream.closed
False
>>> stream.memory is None
False
>>> stream.readable()
True
>>> stream.close()
>>> stream.closed
True
>>> stream.memory is None
True
>>> stream.readable()
Traceback (most recent call last):
    ...
ValueError: I/O operation on closed stream.

property closed: bool

Closed stream.

Returns:

bool – Closed stream.

See also

close()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> stream = MemoryIO(Memory.from_bytes(b'ABC'))
>>> stream.closed
False
>>> stream.close()
>>> stream.closed
True

>>> with MemoryIO(Memory.from_bytes(b'ABC')) as stream:
...     print(stream.closed)
False
>>> print(stream.closed)
True

detach()

Detaches the underlying raw stream.

Warning

It always raises io.UnsupportedOperation. This method is present only for API compatibility. No actual underlying stream is present for this object.

Raises:

io.UnsupportedOperation – No underlying raw stream.

fileno()

File descriptor identifier.

Warning

It always raises io.UnsupportedOperation. This method is present only for API compatibility. No actual file descriptor is associated to this object.

Raises:

OSError – Not a file stream.

flush()

Flushes buffered data into the underlying raw steam.

Notes

Since no underlying stream is associated, this method does nothing.

getbuffer()

Memory view of the underlying memory object.

Warning

This method may fail when the underlying memory object has gaps within data.

Returns:

memoryview – Memory view over the underlying memory object.

See also

ImmutableMemory.view()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> with MemoryIO(Memory.from_bytes(b'Hello, World!')) as stream:
...     with stream.getbuffer() as buffer:
...         print(type(buffer), '=', bytes(buffer))
<class 'memoryview'> = b'Hello, World!'

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> with MemoryIO(Memory.from_blocks(blocks)) as stream:
...     stream.getbuffer()
Traceback (most recent call last):
    ...
ValueError: non-contiguous data within range

getvalue()

Byte string copy of the underlying memory object.

Warning

This method may fail when the underlying memory object has gaps within data.

Returns:

bytes – Byte string copy of the underlying memory object.

See also

ImmutableMemory.to_bytes()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> with MemoryIO(Memory.from_bytes(b'Hello, World!')) as stream:
...     value = stream.getvalue()
...     print(type(value), '=', bytes(value))
<class 'bytes'> = b'Hello, World!'

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> with MemoryIO(Memory.from_blocks(blocks)) as stream:
...     stream.getvalue()
Traceback (most recent call last):
    ...
ValueError: non-contiguous data within range

isatty()

Interactive console stream.

Returns:

bool – False, not an interactive console stream.

property memory: ImmutableMemory | None

Underlying memory object.

None when closed.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> memory = Memory.from_bytes(b'Hello, World!')
>>> with MemoryIO(memory) as stream:
...     print(stream.memory is memory)
True
>>> print(stream.memory is memory)
False
>>> print(stream.memory is None)
True

Type:

ImmutableMemory

peek(size=0, asmemview=False)

Previews the next chunk of bytes.

Similar to read(), without moving the stream position instead. This method can be used to preview the next chunk of bytes, without affecting the stream itself.

The number of returned bytes may be different from size, which acts as a mere hint.

If the current stream position lies within a memory gap, this method returns the negative amount of bytes to reach the next data block.

If the current stream position is after the end of memory data, this method returns an empty byte string.

Parameters:

• size (int) – Number of bytes to read. If negative or None, read as many bytes as possible.

• asmemview (bool) – Return a memoryview instead of bytes.

Returns:

bytes – Chunk of bytes.

See also

read()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)
>>> stream = MemoryIO(memory, seek=4)
>>> stream.peek()
b''
>>> stream.peek(1)
b'e'
>>> stream.peek(11)
b'ello'
>>> stream.peek(None)
b'ello'
>>> stream.tell()
4
>>> memview = stream.peek(-1, asmemview=True)
>>> type(memview)
<class 'memoryview'>
>>> bytes(memview)
b'ello'
>>> stream.seek(8)
8
>>> stream.peek()
-2

read(size=-1, asmemview=False)

Reads a chunk of bytes.

Starting from the current stream position, this method tries to read up to size bytes (or as much as possible if negative or None).

The number of bytes can be less than size in the case a memory hole or the end are encountered.

If the current stream position lies within a memory gap, this method returns the negative amount of bytes to reach the next data block.

If the current stream position is after the end of memory data, this method returns an empty byte string.

Parameters:

• size (int) – Number of bytes to read. If negative or None, read as many bytes as possible.

• asmemview (bool) – Return a memoryview instead of bytes.

Returns:

bytes – Chunk of up to size bytes.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)
>>> stream = MemoryIO(memory, seek=4)
>>> stream.read(1)
b'e'
>>> stream.tell()
5
>>> stream.read(99)
b'llo'
>>> stream.tell()
8
>>> stream.read()
-2
>>> stream.tell()
10
>>> memview = stream.read(None, asmemview=True)
>>> type(memview)
<class 'memoryview'>
>>> bytes(memview)
b'World!'
>>> stream.tell()
16
>>> stream.read()
b''
>>> stream.tell()
16

read1(size=-1, asmemview=False)

Reads a chunk of bytes.

Starting from the current stream position, this method tries to read up to size bytes (or as much as possible if negative or None).

The number of bytes can be less than size in the case a memory hole or the end are encountered.

If the current stream position lies within a memory gap, this method returns the negative amount of bytes to reach the next data block.

If the current stream position is after the end of memory data, this method returns an empty byte string.

Parameters:

• size (int) – Number of bytes to read. If negative or None, read as many bytes as possible.

• asmemview (bool) – Return a memoryview instead of bytes.

Returns:

bytes – Chunk of up to size bytes.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)
>>> stream = MemoryIO(memory, seek=4)
>>> stream.read(1)
b'e'
>>> stream.tell()
5
>>> stream.read(99)
b'llo'
>>> stream.tell()
8
>>> stream.read()
-2
>>> stream.tell()
10
>>> memview = stream.read(None, asmemview=True)
>>> type(memview)
<class 'memoryview'>
>>> bytes(memview)
b'World!'
>>> stream.tell()
16
>>> stream.read()
b''
>>> stream.tell()
16

readable()

Stream is readable.

Returns:

bool – True, stream is always readable.

readinto(buffer, skipgaps=True)

Reads data into a byte buffer.

If the stream is pointing after the memory end, no bytes are read.

If pointing within a memory hole (gap), the negative number of bytes until the next data block is returned. The stream is always positioned after the gap.

If a memory hole (gap) is encountered after reading some bytes, the reading stops there, and the number of bytes read is returned. The stream is always positioned after the gap.

Standard operation reads data until buffer is full, or encountering the memory end. It returns the number of bytes read.

Parameters:

• buffer (bytearray) – A pre-allocated byte array to fill with bytes read from the stream.

• skipgaps (bool) – If false, it stops reading when a memory hole (gap) is encountered.

Returns:

int – Number of bytes read, or the negative gap size.

Examples

>>> from bytesparse import Memory, MemoryIO
>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)

>>> stream = MemoryIO(memory, seek=4)
>>> buffer = bytearray(b'.' * 8)
>>> stream.readinto(buffer, skipgaps=True)
8
>>> buffer
bytearray(b'elloWorl')
>>> stream.tell()
14
>>> stream.readinto(buffer, skipgaps=True)
2
>>> buffer
bytearray(b'd!loWorl')
>>> stream.tell()
16
>>> stream.readinto(buffer, skipgaps=True)
0
>>> buffer
bytearray(b'd!loWorl')
>>> stream.tell()
16

>>> stream = MemoryIO(memory, seek=4)
>>> buffer = bytearray(b'.' * 8)
>>> stream.readinto(buffer, skipgaps=False)
4
>>> buffer
bytearray(b'ello....')
>>> stream.tell()
8
>>> stream.readinto(buffer, skipgaps=False)
-2
>>> stream.tell()
10
>>> stream.readinto(buffer, skipgaps=False)
6
>>> buffer
bytearray(b'World!..')
>>> stream.tell()
16
>>> stream.readinto(buffer, skipgaps=False)
0
>>> buffer
bytearray(b'World!..')
>>> stream.tell()
16

readinto1(buffer, skipgaps=True)

Reads data into a byte buffer.

If the stream is pointing after the memory end, no bytes are read.

If pointing within a memory hole (gap), the negative number of bytes until the next data block is returned. The stream is always positioned after the gap.

If a memory hole (gap) is encountered after reading some bytes, the reading stops there, and the number of bytes read is returned. The stream is always positioned after the gap.

Standard operation reads data until buffer is full, or encountering the memory end. It returns the number of bytes read.

Parameters:

• buffer (bytearray) – A pre-allocated byte array to fill with bytes read from the stream.

• skipgaps (bool) – If false, it stops reading when a memory hole (gap) is encountered.

Returns:

int – Number of bytes read, or the negative gap size.

Examples

>>> from bytesparse import Memory, MemoryIO
>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)

>>> stream = MemoryIO(memory, seek=4)
>>> buffer = bytearray(b'.' * 8)
>>> stream.readinto(buffer, skipgaps=True)
8
>>> buffer
bytearray(b'elloWorl')
>>> stream.tell()
14
>>> stream.readinto(buffer, skipgaps=True)
2
>>> buffer
bytearray(b'd!loWorl')
>>> stream.tell()
16
>>> stream.readinto(buffer, skipgaps=True)
0
>>> buffer
bytearray(b'd!loWorl')
>>> stream.tell()
16

>>> stream = MemoryIO(memory, seek=4)
>>> buffer = bytearray(b'.' * 8)
>>> stream.readinto(buffer, skipgaps=False)
4
>>> buffer
bytearray(b'ello....')
>>> stream.tell()
8
>>> stream.readinto(buffer, skipgaps=False)
-2
>>> stream.tell()
10
>>> stream.readinto(buffer, skipgaps=False)
6
>>> buffer
bytearray(b'World!..')
>>> stream.tell()
16
>>> stream.readinto(buffer, skipgaps=False)
0
>>> buffer
bytearray(b'World!..')
>>> stream.tell()
16

readline(size=-1, skipgaps=True, asmemview=False)

Reads a line.

A standard line is a sequence of bytes terminating with a b'\n' newline character.

If size is provided (not None nor negative), the current line ends there, without a trailing newline character.

If the stream is pointing after the memory end, an empty byte string is returned.

If a memory hole (gap) is encountered, the current line ends there without a trailing newline character. The stream is always positioned after the gap.

If the stream points within a memory hole, it returns the negative number of bytes until the next data block. The stream is always positioned after the gap.

Parameters:

• size (int) – Maximum number of bytes for the line to read. If None or negative, no limit is set.

• skipgaps (bool) – If false, the negative size of the pointed memory hole.

• asmemview (bool) – If true, the returned object is a memoryview instead of bytes.

Returns:

bytes or int – Line read from the stream, or the negative gap size.

See also

read() readlines()

Examples

>>> from bytesparse import Memory, MemoryIO
>>> blocks = [[3, b'Hello\nWorld!'], [20, b'Bye\n'], [28, b'Bye!']]

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.readline()
b'Hello\n'
>>> stream.tell()
9
>>> stream.readline(None)
b'World!'
>>> stream.tell()
15
>>> stream.readline(99)
b'Bye\n'
>>> stream.tell()
24
>>> stream.readline(99)
b'Bye!'
>>> stream.tell()
32
>>> stream.readline()
b''
>>> stream.tell()
32

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.readline(4)
b'Hell'
>>> stream.tell()
7
>>> stream.readline(4)
b'o\n'
>>> stream.tell()
9

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> view = stream.readline(asmemview=True)
>>> type(view) is memoryview
True
>>> bytes(view)
b'Hello\n'

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> # Emulating stream.readlines(skipgaps=False)
>>> lines = []
>>> line = True
>>> while line:
...     line = stream.readline(skipgaps=False)
...     lines.append(line)
>>> lines
[-3, b'Hello\n', b'World!', -5, b'Bye\n', -4, b'Bye!']
>>> stream.tell()
32
>>> stream.readline(skipgaps=False)
b''
>>> stream.tell()
32

readlines(hint=-1, skipgaps=True, asmemview=False)

Reads a list of lines.

It repeatedly calls readline(), collecting the returned values into a list, until the total number of bytes read reaches hint.

If a memory hole (gap) is encountered, the current line ends there without a trailing newline character, and the stream is positioned after the gap.

If skipgaps is false, the list is appended the negative size of each encountered memory hole.

Parameters:

• hint (int) – Number of bytes after which line reading stops. If None or negative, no limit is set.

• skipgaps (bool) – If false, the list hosts the negative size of each memory hole.

• asmemview (bool) – If true, the returned objects are memory views instead of byte strings.

Returns:

list of bytes or int – List of lines and gaps read from the stream.

See also

__iter__() read() readline()

Examples

>>> from bytesparse import Memory, MemoryIO
>>> blocks = [[3, b'Hello\nWorld!'], [20, b'Bye\n'], [28, b'Bye!']]

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.readlines()
[b'Hello\n', b'World!', b'Bye\n', b'Bye!']
>>> stream.tell()
32
>>> stream.readlines()
[]
>>> stream.tell()
32

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.readlines(hint=10)
[b'Hello\n', b'World!']
>>> stream.tell()
15

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> views = stream.readlines(asmemview=True)
>>> all(type(view) is memoryview for view in views)
True
>>> [bytes(view) for view in views]
[b'Hello\n', b'World!', b'Bye\n', b'Bye!']

>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.readlines(skipgaps=False)
[-3, b'Hello\n', b'World!', -5, b'Bye\n', -4, b'Bye!']
>>> stream.tell()
32
>>> stream.readlines(skipgaps=False)
[]
>>> stream.tell()
32

seek(offset, whence=0)

Changes the current stream position.

It performs the classic seek() I/O operation.

The whence can be any of:

• SEEK_SET (0 or None):

  referring to the absolute address 0.

• SEEK_CUR (1):

  referring to the current stream position (tell()).

• SEEK_END (2):

  referring to the memory end (ImmutableMemory.endex).

• SEEK_DATA (3):

  if the current stream position lies within a memory hole, it moves to the beginning of the next data block; no operation is performed otherwise.

• SEEK_HOLE (4):

  if the current stream position lies within a data block, it moves to the beginning of the next memory hole (note: the end of the stream is considered as a memory hole); no operation is performed otherwise.

Parameters:

• offset (int) – Position offset to apply.

• whence (int) – Where the offset is referred. It can be any of the standard SEEK_* values. By default, it refers to the beginning of the stream.

Returns:

int – The updated stream position.

Notes

Stream position is just a number, not related to memory ranges.

Examples

>>> from bytesparse import *

>>> blocks = [[3, b'Hello'], [12, b'World!']]
>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.seek(5)
5
>>> stream.seek(-3, SEEK_END)
15
>>> stream.seek(2, SEEK_CUR)
17
>>> stream.seek(1, SEEK_SET)
1
>>> stream.seek(stream.tell(), SEEK_HOLE)
1
>>> stream.seek(stream.tell(), SEEK_DATA)
3
>>> stream.seek(stream.tell(), SEEK_HOLE)
8
>>> stream.seek(stream.tell(), SEEK_DATA)
12
>>> stream.seek(stream.tell(), SEEK_HOLE)  # EOF
18
>>> stream.seek(stream.tell(), SEEK_DATA)  # EOF
18
>>> stream.seek(22)  # after
22
>>> stream.seek(0)  # before
0

seekable()

Stream is seekable.

Returns:

bool – True, stream is always seekable.

skip_data()

Skips a data block.

It moves the current stream position after the end of the currently pointed data block.

No action is performed if the current stream position lies within a memory hole (gap).

Returns:

int – Updated stream position.

See also

seek()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [12, b'World!']]
>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.skip_data()
0
>>> stream.seek(6)
6
>>> stream.skip_data()
8
>>> stream.skip_data()
8
>>> stream.seek(12)
12
>>> stream.skip_data()
18
>>> stream.skip_data()
18
>>> stream.seek(20)
20
>>> stream.skip_data()
20

skip_hole()

Skips a memory hole.

It moves the current stream position after the end of the currently pointed memory hole (gap).

No action is performed if the current stream position lies within a data block.

Returns:

int – Updated stream position.

See also

seek()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [12, b'World!']]
>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.skip_hole()
3
>>> stream.skip_hole()
3
>>> stream.seek(9)
9
>>> stream.skip_hole()
12
>>> stream.skip_hole()
12
>>> stream.seek(20)
20
>>> stream.skip_hole()
20

tell()

Current stream position.

Returns:

int – Current stream position.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [12, b'World!']]
>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.tell()
0
>>> stream.skip_hole()
3
>>> stream.tell()
3
>>> stream.read(5)
b'Hello'
>>> stream.tell()
8
>>> stream.skip_hole()
12
>>> stream.read()
b'World!'
>>> stream.tell()
18
>>> stream.seek(20)
20
>>> stream.tell()
20

truncate(size=None)

Truncates stream.

If size is provided, it moves the current stream position to it.

Any data after the updated stream position are deleted from the underlying memory object.

The updated stream position can lie outside the actual memory bounds (i.e. extending after the memory). No filling is performed, only the stream position is moved there.

Parameters:

size (int) – If not None, the stream is positioned there.

Returns:

int – Updated stream position.

Raises:

io.UnsupportedOperation – Stream not writable.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [12, b'World!']]
>>> stream = MemoryIO(Memory.from_blocks(blocks))
>>> stream.seek(7)
7
>>> stream.truncate()
7
>>> stream.tell()
7
>>> stream.memory.to_blocks()
[[3, b'Hell']]
>>> stream.truncate(10)
10
>>> stream.tell()
10

>>> memory = Memory.from_bytes(b'Hello, World!')
>>> setattr(memory, 'write', None)  # exception on write()
>>> stream = MemoryIO(memory)
>>> stream.seek(7)
7
>>> stream.truncate()
Traceback (most recent call last):
    ...
io.UnsupportedOperation: truncate

writable()

Stream is writable.

Returns:

bool – Stream is writable.

Examples

>>> from bytesparse import Memory, MemoryIO

>>> memory = Memory.from_bytes(b'Hello, World!')
>>> with MemoryIO(memory) as stream:
...     print(stream.writable())
True
>>> setattr(memory, 'write', None)  # exception on write()
>>> with MemoryIO(memory) as stream:
...     print(stream.writable())
False

write(buffer)

Writes data into the stream.

The behaviour depends on the nature of buffer: byte-like or integer.

Byte-like data are written into the underlying memory object via its bytesparse.base.MutableMemory.write() method, at the current stream position (i.e. tell()). The stream position is always incremented by the size of buffer, regardless of the actual number of bytes written into the underlying memory object (e.g. when cropped by existing bytesparse.base.MutableMemory.bounds_span settings).

If buffer is a positive integer, that is the amount of bytes to bytesparse.base.MutableMemory.clear() from the current stream position onwards. The stream position is incremented by buffer bytes. It returns buffer as a positive number.

If buffer is a negative integer, that is the amount of bytes to bytesparse.base.MutableMemory.delete() from the current stream position onwards. The stream position is not changed. It returns buffer as a positive number.

Notes

buffer is considered an integer if the execution of buffer.__index__() does not raise an Exception.

Parameters:

buffer (bytes) – Byte data to write at the current stream position.

Returns:

int – Size of the written buffer.

Raises:

io.UnsupportedOperation – Stream not writable.

See also

bytesparse.base.MutableMemory.clear() bytesparse.base.MutableMemory.delete() bytesparse.base.MutableMemory.write()

Examples

>>> from bytesparse import Memory, MemoryIO

>>> blocks = [[3, b'Hello'], [10, b'World!']]
>>> memory = Memory.from_blocks(blocks)
>>> stream = MemoryIO(memory, seek=10)
>>> stream.write(b'Human')
5
>>> memory.to_blocks()
[[3, b'Hello'], [10, b'Human!']]
>>> stream.tell()
15
>>> stream.seek(7)
7
>>> stream.write(5)  # clear 5 bytes
5
>>> memory.to_blocks()
[[3, b'Hell'], [12, b'man!']]
>>> stream.tell()
12
>>> stream.seek(7)
7
>>> stream.write(-5)  # delete 5 bytes
5
>>> memory.to_blocks()
[[3, b'Hellman!']]
>>> stream.tell()
7

>>> memory = Memory.from_bytes(b'Hello, World!')
>>> setattr(memory, 'write', None)  # exception on write()
>>> stream = MemoryIO(memory, seek=7)
>>> stream.write(b'Human')
Traceback (most recent call last):
    ...
io.UnsupportedOperation: not writable

writelines(lines)

Writes 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.

If a line is an integer, its behavior is as per write() (positive: clear, negative: delete).

Parameters:

lines (list of bytes) – List of byte strings to write.

See also

bytesparse.base.MutableMemory.clear() bytesparse.base.MutableMemory.delete() write()

Examples

>>> from bytesparse import Memory, MemoryIO
>>> lines = [3, b'Hello\n', b'World!', 5, b'Bye\n', 4, b'Bye!']
>>> stream = MemoryIO()
>>> stream.writelines(lines)
>>> stream.memory.to_blocks()
[[3, b'Hello\nWorld!'], [20, b'Bye\n'], [28, b'Bye!']]