"""
Thin :mod:`ctypes` wrapper around a libretro :term:`core`'s exported ``retro_*`` functions.
.. seealso::
:mod:`libretro.api`
The Python equivalents of the C types passed across this boundary.
:mod:`libretro.session`
The high-level harness that orchestrates calls into a :class:`.Core`.
"""
from __future__ import annotations
from abc import abstractmethod
from collections.abc import Buffer, Callable, Sequence
from copy import deepcopy
from ctypes import (
CDLL,
POINTER,
Array,
byref,
c_bool,
c_char,
c_char_p,
c_int16,
c_size_t,
c_ubyte,
c_uint,
cast,
cdll,
)
from os import PathLike
from typing import Protocol, override
from libretro.api import (
Region,
retro_audio_sample_batch_t,
retro_audio_sample_t,
retro_environment_t,
retro_game_info,
retro_input_poll_t,
retro_input_state_t,
retro_subsystem_info,
retro_system_av_info,
retro_system_info,
retro_video_refresh_t,
)
from libretro.api._utils import memoryview_at
from libretro.api.input import Port
from libretro.ctypes import TypedPointer, c_void_ptr
# TODO: Add a CorePhase enum that's updated when entering/leaving each phase.
# (Some envcalls can only be called in certain phases, so this would be useful for error checking.)
[docs]
class CoreInterface(Protocol):
"""An interface for a libretro core."""
[docs]
@abstractmethod
def set_environment(self, env: retro_environment_t) -> None:
"""Call the core's ``retro_set_environment`` function with the given callback."""
...
[docs]
@abstractmethod
def set_video_refresh(self, video: retro_video_refresh_t) -> None:
"""Call the core's ``retro_set_video_refresh`` function with the given callback."""
...
[docs]
@abstractmethod
def set_audio_sample(self, audio: retro_audio_sample_t | Callable[[int, int], None]) -> None:
"""Call the core's ``retro_set_audio_sample`` function with the given callback."""
...
[docs]
@abstractmethod
def set_audio_sample_batch(self, audio: retro_audio_sample_batch_t) -> None:
"""Call the core's ``retro_set_audio_sample_batch`` function with the given callback."""
...
[docs]
@abstractmethod
def init(self) -> None:
"""Call the core's ``retro_init`` function."""
...
[docs]
@abstractmethod
def deinit(self) -> None:
"""Call the core's ``retro_deinit`` function."""
...
[docs]
@abstractmethod
def api_version(self) -> int:
"""Call the core's ``retro_api_version`` function and return the result."""
...
[docs]
@abstractmethod
def get_system_info(self) -> retro_system_info:
"""Call the core's ``retro_get_system_info`` function and return the result."""
...
[docs]
@abstractmethod
def get_system_av_info(self) -> retro_system_av_info:
"""Call the core's ``retro_get_system_av_info`` function and return the result."""
...
[docs]
@abstractmethod
def set_controller_port_device(self, port: Port, device: int) -> None:
"""Call the core's ``retro_set_controller_port_device`` function with the given arguments."""
...
[docs]
@abstractmethod
def reset(self) -> None:
"""Call the core's ``retro_reset`` function."""
...
[docs]
@abstractmethod
def run(self) -> None:
"""Call the core's ``retro_run`` function to advance the core by one frame."""
...
[docs]
@abstractmethod
def serialize_size(self) -> int:
"""Call the core's ``retro_serialize_size`` function and return the result."""
...
[docs]
@abstractmethod
def serialize(self, data: bytearray | memoryview[int]) -> bool:
"""Call the core's ``retro_serialize`` function with the given mutable buffer."""
...
[docs]
@abstractmethod
def unserialize(self, data: bytes | bytearray | memoryview[int]) -> bool:
"""Call the core's ``retro_unserialize`` function with the given buffer."""
...
[docs]
@abstractmethod
def cheat_reset(self) -> None:
"""Call the core's ``retro_cheat_reset`` function."""
...
[docs]
@abstractmethod
def cheat_set(self, index: int, enabled: bool, code: bytes | bytearray | str) -> None:
"""Call the core's ``retro_cheat_set`` function with the given arguments."""
...
[docs]
@abstractmethod
def load_game(self, game: retro_game_info | None) -> bool:
"""Call the core's ``retro_load_game`` function with the given game info."""
...
[docs]
@abstractmethod
def load_game_special(self, game_type: int, info: Sequence[retro_game_info]) -> bool:
"""Call the core's ``retro_load_game_special`` function with the given arguments."""
...
[docs]
@abstractmethod
def unload_game(self) -> None:
"""Call the core's ``retro_unload_game`` function."""
...
[docs]
@abstractmethod
def get_region(self) -> Region | int:
"""Call the core's ``retro_get_region`` function and return the result."""
...
[docs]
@abstractmethod
def get_memory_data(self, id: int) -> c_void_ptr | None:
"""Call the core's ``retro_get_memory_data`` function for the given memory region."""
...
[docs]
@abstractmethod
def get_memory_size(self, id: int) -> int:
"""Call the core's ``retro_get_memory_size`` function for the given memory region."""
...
[docs]
def get_memory(self, id: int) -> memoryview | None:
"""
Get a writable ``memoryview`` of the memory region given by ``id``.
Convenience wrapper around :meth:`get_memory_data` and :meth:`get_memory_size`.
:param id: The ID of the memory region to access.
:return: A writable ``memoryview`` of the given region,
or ``None`` if ``retro_get_memory_data`` returned ``NULL``.
"""
data = self.get_memory_data(id)
if not data:
return None
size = self.get_memory_size(id)
return memoryview_at(data, size, readonly=False)
[docs]
class Core(CoreInterface):
"""
A thin wrapper around a libretro core that can be used to call its public interface.
Does not manage the underlying core's life cycle,
i.e. ``retro_*`` methods are not called implicitly unless otherwise noted;
that's left to ``Session`` or some custom abstraction layer.
"""
[docs]
def __init__(self, core: CDLL | PathLike[str] | PathLike[bytes] | str):
"""
Create a new ``Core`` instance.
:param core: The core to wrap. Can be one of the following:
- A ``str`` or ``PathLike`` representing the path to the core's shared library.
- A ``CDLL`` representing the core's shared library.
:raises ValueError: If the core does not define all the required functions
(i.e. the ``retro_*`` function that each method corresponds to).
:raises TypeError: If ``core`` is not one of the above-mentioned types.
"""
match core:
case CDLL():
self._core = core
case (str() | PathLike()) as path:
self._core = cdll.LoadLibrary(str(path))
case _:
raise TypeError(
f"Expected a CDLL instance or a path to a core, got {type(core).__name__}"
)
try:
self._core.retro_set_environment.argtypes = [retro_environment_t]
self._core.retro_set_environment.restype = None
self._core.retro_set_video_refresh.argtypes = [retro_video_refresh_t]
self._core.retro_set_video_refresh.restype = None
self._core.retro_set_audio_sample.argtypes = [retro_audio_sample_t]
self._core.retro_set_audio_sample.restype = None
self._core.retro_set_audio_sample_batch.argtypes = [retro_audio_sample_batch_t]
self._core.retro_set_audio_sample_batch.restype = None
self._core.retro_set_input_poll.argtypes = [retro_input_poll_t]
self._core.retro_set_input_poll.restype = None
self._core.retro_set_input_state.argtypes = [retro_input_state_t]
self._core.retro_set_input_state.restype = None
self._core.retro_init.argtypes = []
self._core.retro_init.restype = None
self._core.retro_deinit.argtypes = []
self._core.retro_deinit.restype = None
self._core.retro_api_version.argtypes = []
self._core.retro_api_version.restype = c_uint
self._core.retro_get_system_info.argtypes = [
POINTER(retro_system_info),
]
self._core.retro_get_system_info.restype = None
self._core.retro_get_system_av_info.argtypes = [
POINTER(retro_system_av_info),
]
self._core.retro_get_system_av_info.restype = None
self._core.retro_set_controller_port_device.argtypes = [c_uint, c_uint]
self._core.retro_set_controller_port_device.restype = None
self._core.retro_reset.argtypes = []
self._core.retro_reset.restype = None
self._core.retro_run.argtypes = []
self._core.retro_run.restype = None
self._core.retro_serialize_size.argtypes = []
self._core.retro_serialize_size.restype = c_size_t
self._core.retro_serialize.argtypes = [c_void_ptr, c_size_t]
self._core.retro_serialize.restype = c_bool
self._core.retro_unserialize.argtypes = [c_void_ptr, c_size_t]
self._core.retro_unserialize.restype = c_bool
self._core.retro_cheat_reset.argtypes = []
self._core.retro_cheat_reset.restype = None
self._core.retro_cheat_set.argtypes = [c_uint, c_bool, c_char_p]
self._core.retro_cheat_set.restype = None
self._core.retro_load_game.argtypes = [POINTER(retro_game_info)]
self._core.retro_load_game.restype = c_bool
self._core.retro_load_game_special.argtypes = [
c_uint,
POINTER(retro_game_info),
c_size_t,
]
self._core.retro_load_game_special.restype = c_bool
self._core.retro_unload_game.argtypes = []
self._core.retro_unload_game.restype = None
self._core.retro_get_region.argtypes = []
self._core.retro_get_region.restype = c_uint
self._core.retro_get_memory_data.argtypes = [c_uint]
self._core.retro_get_memory_data.restype = POINTER(c_ubyte)
self._core.retro_get_memory_data.errcheck = lambda result, _func, _args: (
cast(result, c_void_ptr) if result else c_void_ptr()
)
self._core.retro_get_memory_size.argtypes = [c_uint]
self._core.retro_get_memory_size.restype = c_size_t
except AttributeError as e:
raise ValueError(
f"Couldn't find required symbol '{e.name}' in {self._core._name}"
) from e
# Need to keep references to these objects to prevent them from being garbage collected,
# otherwise the C function pointers to them will become invalid.
self._environment: retro_environment_t | None = None
self._video_refresh: retro_video_refresh_t | None = None
self._audio_sample: retro_audio_sample_t | None = None
self._audio_sample_batch: retro_audio_sample_batch_t | None = None
self._input_poll: retro_input_poll_t | None = None
self._input_state: retro_input_state_t | None = None
[docs]
@override
def set_environment(
self, env: retro_environment_t | Callable[[int, c_void_ptr], bool]
) -> None:
"""
Call the core's ``retro_set_environment`` function with the given callback.
:param env: The function that the core should use for environment calls.
:raises TypeError: If ``env`` is not a ``retro_environment_t``.
"""
self._environment = retro_environment_t(env)
self._core.retro_set_environment(self._environment)
[docs]
@override
def set_video_refresh(
self, video: retro_video_refresh_t | Callable[[c_void_ptr, int, int, int], None]
) -> None:
"""
Call the core's ``retro_set_video_refresh`` function with the given callback.
:param video: The function that the core should call to update its video output.
:raises TypeError: If ``video`` is not a ``retro_video_refresh_t``.
"""
self._video_refresh = retro_video_refresh_t(video)
self._core.retro_set_video_refresh(self._video_refresh)
[docs]
@override
def set_audio_sample(self, audio: retro_audio_sample_t | Callable[[int, int], None]) -> None:
"""
Call the core's ``retro_set_audio_sample`` function with the given callback.
:param audio: The function that the core should call to render a single audio frame.
:raises TypeError: If ``audio`` is not a ``retro_audio_sample_t``.
"""
self._audio_sample = retro_audio_sample_t(audio)
self._core.retro_set_audio_sample(self._audio_sample)
[docs]
@override
def set_audio_sample_batch(
self, audio: retro_audio_sample_batch_t | Callable[[TypedPointer[c_int16], int], int]
) -> None:
"""
Call the core's ``retro_set_audio_sample_batch`` function with the given callback.
:param audio: The function that the core should call to render a batch of audio frames.
:raises TypeError: If ``audio`` is not a ``retro_audio_sample_batch_t``.
"""
self._audio_sample_batch = retro_audio_sample_batch_t(audio)
self._core.retro_set_audio_sample_batch(self._audio_sample_batch)
[docs]
@override
def init(self):
"""
Call the core's ``retro_init`` function.
:note: This method does not check if the core has already been initialized.
Additionally, this method is not implicitly called by ``__init__``.
"""
self._core.retro_init()
[docs]
@override
def deinit(self):
"""
Call the core's ``retro_deinit`` function.
:note: This method does not validate that the core has been initialized.
Additionally, it is not implicitly called upon deletion.
"""
self._core.retro_deinit()
[docs]
@override
def api_version(self) -> int:
"""
Call the core's ``retro_api_version`` function.
:return: The integer returned by the core's implementation of ``retro_api_version``.
:warning: This method does not validate the returned version number.
"""
return self._core.retro_api_version()
[docs]
@override
def get_system_info(self) -> retro_system_info:
"""
Call the core's ``retro_get_system_info`` function.
:return: A ``retro_system_info`` instance containing information about the core.
All strings are copied and may be accessed even after unloading the core.
"""
system_info = retro_system_info()
self._core.retro_get_system_info(byref(system_info))
return deepcopy(system_info)
[docs]
@override
def get_system_av_info(self) -> retro_system_av_info:
"""
Call the core's ``retro_get_system_av_info`` function.
:return: A ``retro_system_av_info`` instance
containing information about the core's audiovisual capabilities.
It may be accessed even after unloading the core.
"""
system_av_info = retro_system_av_info()
self._core.retro_get_system_av_info(byref(system_av_info))
return system_av_info
[docs]
@override
def set_controller_port_device(self, port: Port, device: int):
"""
Call the core's ``retro_set_controller_port_device`` function with the given arguments.
:param port: The port to set the device for.
Masked to fit within the range of an ``unsigned int``.
:param device: The device to assign to ``port``.
Masked to fit within the range of an ``unsigned int``.
"""
self._core.retro_set_controller_port_device(port, device)
[docs]
@override
def reset(self):
"""
Call the core's ``retro_reset`` function.
:warning: Does not check if the core is in a state where it can be reset.
"""
self._core.retro_reset()
[docs]
@override
def run(self):
"""
Call the core's ``retro_run`` function.
:warning: Does not check if the core is in a state where it can be run.
"""
self._core.retro_run()
[docs]
@override
def serialize_size(self) -> int:
"""
Call the core's ``retro_serialize_size`` function.
:return: The length of the buffer needed to serialize the core's state, in bytes.
If zero, the core does not support serialization.
"""
return self._core.retro_serialize_size()
[docs]
@override
def serialize(self, data: bytearray | memoryview[int] | Buffer) -> bool:
"""
Call the core's ``retro_serialize`` function with the given mutable buffer.
Fills ``data`` with whatever serialized state the core returns.
:param data: A ``bytearray``, mutable ``memoryview``, or ``Buffer`` implementation
that core's serialized state will be saved to.
:return: ``True`` if the core successfully serialized its state, ``False`` otherwise.
:raise TypeError: If ``data`` is not one of the aforementioned types.
:raise ValueError: If ``data`` is a read-only ``memoryview`` or ``Buffer``.
:note: The buffer must be at least as large as the last value returned by ``serialize_size``,
or else the serialized data will be incomplete.
"""
match data:
case memoryview() if data.readonly:
raise ValueError("data must not be readonly")
case memoryview():
buf = data.cast("B")
case bytearray() | Buffer():
buf = memoryview(data)
case _:
raise TypeError(
f"Expected a bytearray, writable Buffer, or writable memoryview; got {type(data).__name__}"
)
buflength = len(buf)
arraytype = c_char * buflength
return self._core.retro_serialize(byref(arraytype.from_buffer(buf)), buflength)
[docs]
@override
def unserialize(self, data: bytes | bytearray | memoryview[int] | Buffer) -> bool:
"""
Call the core's ``retro_unserialize`` function with the given buffer.
Restores the core's state from the serialized data in ``data``.
:param data: A ``bytes``, ``bytearray``, ``memoryview``, or ``Buffer``.
:raises TypeError: If ``data`` is not one of the aforementioned types.
:return: ``True`` if the core successfully loaded a state from ``data``, ``False`` if not.
"""
match data:
case bytes():
buf = memoryview_at(data, len(data), readonly=False)
# HACK! ctypes.Array.from_buffer requires a writable buffer,
# but bytes objects are read-only.
# retro_unserialize isn't supposed to modify the buffer,
# so we can blame undefined behavior if the core tries to write to it anyway.
case memoryview():
buf = data.cast("B")
case bytearray() | Buffer():
buf = memoryview(data)
case _:
raise TypeError(
f"Expected bytes, bytearray, memoryview, or Buffer; got {type(data).__name__}"
)
buflen = len(buf)
arraytype = c_char * buflen
# TODO: Validate that the buffer wasn't written to, and raise a warning if it was. (Use zlib.crc32)
return self._core.retro_unserialize(byref(arraytype.from_buffer(buf)), buflen)
[docs]
@override
def cheat_reset(self):
"""Call the core's ``retro_cheat_reset`` function."""
self._core.retro_cheat_reset()
[docs]
@override
def cheat_set(self, index: int, enabled: bool, code: bytes | bytearray | str):
"""
Call the core's ``retro_cheat_set`` function with the given arguments.
:param index: The number of the cheat code to toggle.
:param enabled: Whether the cheat code should be enabled or disabled.
:param code: A buffer containing a zero-terminated byte string.
:raise TypeError: If any parameter's value is inconsistent with its documented type.
:raise ValueError: If ``code`` does not contain a null terminator (i.e. the value 0).
"""
if not isinstance(index, int):
raise TypeError(f"Expected int, got {type(index).__name__}")
if not isinstance(enabled, bool):
raise TypeError(f"Expected bool, got {type(enabled).__name__}")
buf: bytes
match code:
case bytes():
buf = code
case bytearray():
buf = bytes(code)
case str():
buf = code.encode()
case _:
raise TypeError(f"Expected bytes, bytearray, or str; got {type(code).__name__}")
self._core.retro_cheat_set(index, enabled, buf)
[docs]
@override
def load_game(self, game: retro_game_info | None) -> bool:
"""
Call the core's ``retro_load_game`` function with the given game info.
:param game: A ``retro_game_info`` instance or ``None``.
:return: ``True`` if the core successfully loaded ``game``, ``False`` otherwise.
:raises TypeError: If ``game`` is not a ``retro_game_info`` or ``None``.
:warning: This method does not validate any preconditions documented in libretro.h,
e.g. it's possible to pass ``None`` even if the core doesn't support no-content mode.
"""
match game:
case retro_game_info():
return self._core.retro_load_game(byref(game))
case None:
return self._core.retro_load_game(None)
case _:
raise TypeError(f"Expected retro_game_info or None, got {type(game).__name__}")
[docs]
@override
def load_game_special(
self,
game_type: int | retro_subsystem_info,
info: Sequence[retro_game_info] | Array[retro_game_info],
) -> bool:
"""
Call the core's ``retro_load_game_special`` function with the given arguments.
:param game_type: The subsystem type to activate.
May be passed as an ``int`` or a ``retro_subsystem_info`` instance.
:param info: A ``Sequence`` or ``ctypes.Array`` of ``retro_game_info`` instances.
:return: ``True`` if the core successfully loaded the game, ``False`` if not.
:raises TypeError: If any parameter's value is inconsistent with its documented type.
:warning: This method does not validate any preconditions documented in libretro.h,
e.g. it's possible to use this method even if the core doesn't define subsystems.
"""
match game_type:
case int():
_type = game_type
case retro_subsystem_info():
_type = game_type.id
case _:
raise TypeError(
f"Expected int or retro_subsystem_info, got {type(game_type).__name__}"
)
match info:
case Array():
info_array = info
case Sequence():
GameInfoArray = retro_game_info * len(info)
info_array = GameInfoArray(*info)
case _:
raise TypeError(
f"Expected a Sequence or ctypes Array of retro_game_info, got {type(info).__name__}"
)
return self._core.retro_load_game_special(_type, info_array, len(info_array))
[docs]
@override
def unload_game(self):
"""
Call the core's ``retro_unload_game`` function.
:warning: Does not check if the preconditions for ``retro_unload_game`` are met,
e.g. it doesn't check if a game is currently loaded.
"""
self._core.retro_unload_game()
[docs]
@override
def get_region(self) -> Region | int:
"""
Call the core's ``retro_get_region`` function.
:return: The returned region as a ``Region`` enum if it's a known value,
or as a plain ``int`` if not.
"""
region: int = self._core.retro_get_region()
return Region(region) if region in Region else region
[docs]
@override
def get_memory_data(self, id: int) -> c_void_ptr | None:
"""
Call the core's ``retro_get_memory_data`` function for the given memory region.
:param id: The ID of the memory region to access.
:return: Pointer to the memory region returned by the core,
or ``None`` if the core returned ``NULL``.
:raises TypeError: If ``id`` is not an ``int``.
"""
if not isinstance(id, int):
raise TypeError(f"Expected int, got {type(id).__name__}")
return self._core.retro_get_memory_data(id)
[docs]
@override
def get_memory_size(self, id: int) -> int:
"""
Call the core's ``retro_get_memory_size`` function for the given memory region.
:param id: The ID of the memory region to get the size of.
:raises TypeError: If ``id`` is not an ``int``.
:return: The size of the memory region, in bytes.
"""
if not isinstance(id, int):
raise TypeError(f"Expected int, got {type(id).__name__}")
return self._core.retro_get_memory_size(id)
@property
def path(self) -> str:
"""The path to the core's shared library."""
return self._core._name
__all__ = [
"CoreInterface",
"Core",
]