Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions PyMemoryEditor/linux/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,6 +30,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -378,6 +379,18 @@ def read_process_memory(pid: int, address: int, pytype: Type[T], bufflength: int
return data.value


def read_process_memory_into(pid: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``_LinuxPartialIOError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _process_vm_readv(pid, addressof(c_buffer), address, size)


def search_addresses_by_value(
pid: int,
pytype: Type[T],
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/linux/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,6 +21,7 @@
get_modules,
get_threads,
read_process_memory,
read_process_memory_into,
search_addresses_by_pattern,
search_addresses_by_value,
search_values_by_addresses,
Expand DownExpand Up@@ -117,6 +118,10 @@ def read_process_memory(
self.pid, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.pid, address, buffer)

def search_by_addresses(
self,
pytype: Type[T],
Expand Down
13 changes: 13 additions & 0 deletions PyMemoryEditor/macos/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -28,6 +28,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -545,6 +546,18 @@ def read_process_memory(
return data.value


def read_process_memory_into(task: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read (always the buffer's byte length on success; a short read raises
``MachPartialReadError``).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
return _mach_read(task, address, ctypes.addressof(c_buffer), size)


def write_process_memory(
task: int,
address: int,
Expand Down
5 changes: 5 additions & 0 deletions PyMemoryEditor/macos/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,6 +25,7 @@
get_task_for_pid,
get_threads,
read_process_memory,
read_process_memory_into,
release_task,
search_addresses_by_pattern,
search_addresses_by_value,
Expand DownExpand Up@@ -292,6 +293,10 @@ def read_process_memory(
self.__task, address, pytype, resolve_bufflength(pytype, bufflength)
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
return read_process_memory_into(self.__task, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
43 changes: 43 additions & 0 deletions PyMemoryEditor/process/abstract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -449,6 +449,49 @@ def read_process_memory(
"""
raise NotImplementedError()

@abstractmethod
def read_process_memory_into(self, address: int, buffer: object) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` and return the number of bytes read.

This is the zero-copy counterpart of :meth:`read_process_memory`: where
that method allocates a fresh object on every call, this one fills a
buffer you own and reuse. In a tight polling / recording loop that reads
the same-sized region over and over, reusing one buffer keeps memory use
constant instead of producing a stream of short-lived objects for the
garbage collector to reclaim.

:param address: target memory address (ex: 0x006A9EC0).
:param buffer: any writable, contiguous buffer-protocol object — a
``bytearray``, a ``ctypes`` array, a writable ``memoryview``, a
``numpy`` array, etc. Its byte length determines how many bytes are
read; an element-typed buffer (e.g. a ``numpy`` ``int32`` array) is
sized in **bytes**, not elements. The bytes are written in place; no
decoding is performed (this is the raw-``bytes`` read path). Decode
or reinterpret them yourself afterwards
(``int.from_bytes`` / ``struct.unpack`` / ``numpy`` views / ...).
:return: the number of bytes read — always equal to the buffer's byte
length on success (a short read raises instead, mirroring
:meth:`read_process_memory`).

:raises TypeError: if ``buffer`` is not a writable buffer (e.g. an
immutable ``bytes`` object).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails, or returns fewer bytes than
requested (e.g. the range crosses an unreadable/freed page).

Example
-------
::

buffer = bytearray(16)
while recording:
process.read_process_memory_into(addr, buffer)
handle(buffer) # same buffer reused every iteration
"""
raise NotImplementedError()

@abstractmethod
def write_process_memory(
self,
Expand Down
1 change: 1 addition & 0 deletions PyMemoryEditor/util/__init__.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,7 @@
UNSET,
_check_int_fits,
_validate_pytype,
as_writable_c_buffer,
convert_from_byte_array,
get_c_type_of,
prepare_write,
Expand Down
48 changes: 48 additions & 0 deletions PyMemoryEditor/util/convert.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -209,6 +209,54 @@ def prepare_write(
return pytype, length, value


def as_writable_c_buffer(buffer: Any) -> "ctypes.Array":
"""
Wrap a writable buffer-protocol object as a ``ctypes`` ``c_char`` array
that shares the same underlying storage, so a backend can read process
memory *directly into the caller's buffer* with no intermediate
allocation. Returning the ``ctypes`` array (rather than a bare address)
keeps a reference to the source alive for as long as the caller holds it,
which is what makes the address safe to pass to the OS read call.

Accepts anything exposing a writable, contiguous buffer — ``bytearray``,
a ``ctypes`` array, a writable ``memoryview``, a ``numpy`` array, etc. The
array's byte length is taken from the buffer itself (``memoryview.nbytes``),
so element-typed buffers like a ``numpy`` ``int32`` array are sized in
bytes, not elements.

:raises TypeError: if ``buffer`` does not support the buffer protocol or is
read-only (e.g. ``bytes`` — use ``bytearray`` instead).
:raises ValueError: if ``buffer`` is empty or not contiguous.
"""
try:
view = memoryview(buffer)
except TypeError:
raise TypeError(
"buffer must support the writable buffer protocol "
"(e.g. bytearray, a ctypes array, or a numpy array), got %s."
% type(buffer).__name__
)

try:
if view.readonly:
raise TypeError(
"buffer must be writable; got a read-only buffer "
"(e.g. bytes). Use bytearray or another writable buffer."
)
if not view.contiguous:
raise ValueError("buffer must be contiguous.")
nbytes = view.nbytes
finally:
# Release the inspection view promptly so it never lingers as an extra
# export on the source object (e.g. blocking a later bytearray resize).
view.release()

if nbytes == 0:
raise ValueError("buffer must have a non-zero length.")

return (ctypes.c_char * nbytes).from_buffer(buffer)


def convert_from_byte_array(
byte_array: ctypes.Array, pytype: Type[T], length: int
) -> T:
Expand Down
35 changes: 35 additions & 0 deletions PyMemoryEditor/win32/functions.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
from ..process.thread_info import ThreadInfo
from ..util import (
_validate_pytype,
as_writable_c_buffer,
get_c_type_of,
values_to_bytes,
)
Expand DownExpand Up@@ -501,6 +502,40 @@ def ReadProcessMemory(
return data.value


def ReadProcessMemoryInto(process_handle: int, address: int, buffer) -> int:
"""
Read ``len(buffer)`` bytes from ``address`` directly into the writable
``buffer``, with no intermediate allocation. Returns the number of bytes
read.

Raises OSError if the read fails or returns fewer bytes than requested
(same partial-read guard as :func:`ReadProcessMemory`).
"""
c_buffer = as_writable_c_buffer(buffer)
size = len(c_buffer)
bytes_read = ctypes.c_size_t(0)

ctypes.set_last_error(0)
success = kernel32.ReadProcessMemory(
process_handle,
ctypes.c_void_p(address),
ctypes.byref(c_buffer),
size,
ctypes.byref(bytes_read),
)

if not success:
_raise_last_error("ReadProcessMemory")

if bytes_read.value != size:
raise OSError(
"ReadProcessMemory partial read at 0x%X: %d of %d bytes read."
% (address, bytes_read.value, size)
)

return bytes_read.value


def _read_region(process_handle: int, address: int, size: int):
"""Read a memory region; returns the byte buffer or None on failure."""
region_data = (ctypes.c_byte * size)()
Expand Down
6 changes: 6 additions & 0 deletions PyMemoryEditor/win32/process.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,6 +27,7 @@
GetProcessHandle,
GetThreads,
ReadProcessMemory,
ReadProcessMemoryInto,
SearchAddressesByPattern,
SearchAddressesByValue,
SearchValuesByAddresses,
Expand DownExpand Up@@ -313,6 +314,11 @@ def read_process_memory(
resolve_bufflength(pytype, bufflength),
)

def read_process_memory_into(self, address: int, buffer: object) -> int:
self.__require_open()
self.__require_read()
return ReadProcessMemoryInto(self.__process_handle, address, buffer)

def write_process_memory(
self,
address: int,
Expand Down
17 changes: 17 additions & 0 deletions docs/api/openprocess.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -158,6 +158,23 @@ with OpenProcess(
:returns: the original ``value`` you passed in — **not** the truncated/encoded
form actually written (a capped ``str``/``bytes`` write returns the full
original value).

.. py:method:: read_process_memory_into(address, buffer)

Read ``len(buffer)`` raw bytes from ``address`` directly into a
pre-allocated, writable ``buffer`` (no intermediate allocation) — the
zero-copy counterpart of :py:meth:`read_process_memory` for tight
read-the-same-region loops. See :doc:`../guide/read-write` for examples.

:param int address: target memory address.
:param buffer: any writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes land
verbatim (no decoding).
:returns: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

### Typed shortcuts
Expand Down
54 changes: 54 additions & 0 deletions docs/guide/read-write.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -195,6 +195,60 @@ Need the raw bytes with zero interpretation? Use `read_bytes(address, length)`
and `write_bytes(address, data)`.
```

## Reusing a buffer (zero-copy reads)

Every `read_process_memory` call **allocates a fresh Python object** for the
result. That's fine for one-off reads, but in a tight loop that reads the same
region thousands of times — a recorder, a live overlay, a poller — those
throwaway objects pile up and keep the garbage collector busy.

`read_process_memory_into(address, buffer)` reads straight into a buffer **you
own and reuse**, so a long-running loop runs with constant memory instead of a
steady stream of allocations:

```python
buffer = bytearray(16) # allocate once

with OpenProcess(name="game.exe") as process:
while recording:
process.read_process_memory_into(0x7FF40010, buffer)
handle(buffer) # same buffer, refilled in place every loop
```

It fills **`len(buffer)` bytes** (the buffer's size decides how much is read)
and returns the number of bytes read. The bytes land verbatim — no decoding —
so reinterpret them yourself with `int.from_bytes`, `struct.unpack`, a `numpy`
view, and so on.

Any writable, contiguous buffer works — a `bytearray`, a `ctypes` array, a
writable `memoryview`, or a `numpy` array (sized in **bytes**, so a 4-element
`int32` array reads 16 bytes):

```python
import numpy as np

frame = np.zeros(4, dtype=np.int32) # 16 bytes
process.read_process_memory_into(0x7FF40010, frame)
# frame now holds the four int32 values, no per-read allocation
```

### Method signature

```{eval-rst}
.. py:method:: read_process_memory_into(address, buffer)
:no-index:

:param int address: target memory address.
:param buffer: a writable, contiguous buffer-protocol object
(``bytearray``, ``ctypes`` array, writable ``memoryview``, ``numpy``
array, …). Its byte length sets how many bytes are read; the bytes are
written in place with no decoding.
:return: the number of bytes read (the buffer's byte length on success).
:raises TypeError: if ``buffer`` is not a writable buffer (e.g. ``bytes``).
:raises ValueError: if ``buffer`` is empty or not contiguous.
:raises OSError: if the read fails or returns fewer bytes than requested.
```

## Common errors

- **`OSError`** — the address may have been freed between scan and write, or
Expand Down
Loading
Loading