API reference

mister_fpga.MisterClient

Async REST client for the MiSTer FPGA mrext Remote API.

Wraps every endpoint exposed by the mrext Remote service under http://host:port/api. All network I/O is non-blocking; methods raise :class:MisterConnectionError on any network or HTTP error.

Parameters:

Name	Type	Description	Default
host	str	IP address or hostname of the MiSTer device.	required
port	int	mrext Remote port (default 8182).	8182
session	ClientSession | None	Optional shared aiohttp.ClientSession. When None the client creates and owns its own session; call :meth:async_close to release it.	None
timeout	int	Per-request timeout in seconds (default 10).	10

Source code in src/mister_fpga/client.py

class MisterClient:
    """Async REST client for the MiSTer FPGA mrext Remote API.

    Wraps every endpoint exposed by the mrext Remote service under
    ``http://host:port/api``.  All network I/O is non-blocking; methods raise
    :class:`MisterConnectionError` on any network or HTTP error.

    Args:
        host: IP address or hostname of the MiSTer device.
        port: mrext Remote port (default ``8182``).
        session: Optional shared ``aiohttp.ClientSession``.  When ``None`` the
            client creates and owns its own session; call
            :meth:`async_close` to release it.
        timeout: Per-request timeout in seconds (default ``10``).
    """

    def __init__(
        self,
        host: str,
        port: int = 8182,
        *,
        session: aiohttp.ClientSession | None = None,
        timeout: int = 10,
    ) -> None:
        self.host = host
        self.port = port
        self.base_url = f"http://{host}:{port}/api"
        self._timeout = timeout
        self._session = session
        self._owns_session = session is None

    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None:
            self._session = aiohttp.ClientSession()
        return self._session

    async def async_close(self) -> None:
        """Close the underlying HTTP session if this client created it."""
        if self._owns_session and self._session is not None:
            await self._session.close()
            self._session = None

    async def _request(
        self,
        method: str,
        path: str,
        *,
        payload: dict | None = None,
        parse_json: bool = True,
    ) -> Any:
        session = await self._get_session()
        url = f"{self.base_url}{path}"
        try:
            async with session.request(
                method,
                url,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=self._timeout),
            ) as resp:
                resp.raise_for_status()
                if not parse_json:
                    return await resp.read()
                text = await resp.text()
                if not text.strip():
                    return None
                try:
                    return json.loads(text)
                except json.JSONDecodeError as err:
                    raise MisterConnectionError(
                        f"{method} {url} returned invalid JSON: {err}"
                    ) from err
        except (TimeoutError, aiohttp.ClientError) as err:
            _LOGGER.debug("Request %s %s failed: %s", method, url, err)
            raise MisterConnectionError(f"{method} {url} failed: {err}") from err

    async def async_get_status(self) -> MisterStatus:
        """Fetch a combined sysinfo + playing snapshot.

        Returns:
            A :class:`MisterStatus` with ``online=True`` on success.

        Raises:
            MisterConnectionError: If either underlying request fails.
        """
        sysinfo = await self._request("GET", PATH_SYSINFO) or {}
        playing = await self._request("GET", PATH_PLAYING) or {}
        ips = sysinfo.get("ips") or []
        disks = sysinfo.get("disks") or []
        disk = disks[0] if disks else {}
        return MisterStatus(
            online=True,
            core=playing.get("core") or None,
            system=playing.get("system") or None,
            system_name=playing.get("systemName") or None,
            game=playing.get("game") or None,
            game_name=playing.get("gameName") or None,
            hostname=sysinfo.get("hostname"),
            version=sysinfo.get("version"),
            ips=ips,
            ip=ips[0] if ips else None,
            updated=sysinfo.get("updated"),
            dns=sysinfo.get("dns"),
            disk_total=disk.get("total"),
            disk_used=disk.get("used"),
            disk_free=disk.get("free"),
        )

    async def async_get_systems(self) -> list[dict]:
        """Return the list of systems known to mrext."""
        return await self._request("GET", PATH_SYSTEMS) or []

    async def async_launch_system(self, system_id: str) -> None:
        """Launch the default game/core for *system_id*."""
        await self._request("POST", f"{PATH_SYSTEMS}/{system_id}")

    async def async_launch_game(self, path: str) -> None:
        """Launch a game by its absolute path on the MiSTer SD card."""
        await self._request("POST", PATH_GAMES_LAUNCH, payload={"path": path})

    async def async_launch_menu(self) -> None:
        """Return to the MiSTer main menu."""
        await self._request("POST", PATH_LAUNCH_MENU)

    async def async_search_games(self, query: str, system: str = "all") -> dict:
        """Search the game index.

        Args:
            query: Free-text search string.
            system: Limit results to a system slug, or ``"all"`` (default).

        Returns:
            A dict with a ``results`` key containing matching game entries.
        """
        return (
            await self._request(
                "POST", PATH_GAMES_SEARCH, payload={"data": query, "system": system}
            )
            or {}
        )

    async def async_index_games(self) -> None:
        """Trigger a background re-index of the game library."""
        await self._request("POST", PATH_GAMES_INDEX)

    async def async_send_keyboard(self, name: str) -> None:
        """Send a named virtual-keyboard event (see ``KEYBOARD_NAMES``)."""
        await self._request("POST", f"{PATH_KEYBOARD}/{name}")

    async def async_reboot(self) -> None:
        """Reboot the MiSTer device."""
        await self._request("POST", PATH_REBOOT)

    async def async_restart_remote(self) -> None:
        """Restart the mrext Remote service on the MiSTer."""
        await self._request("POST", PATH_RESTART_REMOTE)

    async def async_take_screenshot(self) -> None:
        """Capture a screenshot on the MiSTer and save it server-side."""
        await self._request("POST", PATH_SCREENSHOTS)

    async def async_get_screenshots(self) -> list[dict]:
        """Return metadata for all saved screenshots."""
        return await self._request("GET", PATH_SCREENSHOTS) or []

    async def async_get_screenshot_image(self, core: str, filename: str) -> bytes:
        """Download a screenshot image as raw bytes.

        Args:
            core: Core name sub-directory (e.g. ``"SNES"``).
            filename: Screenshot filename within that sub-directory.

        Returns:
            Raw image bytes (typically PNG).

        Raises:
            MisterConnectionError: If the image cannot be retrieved.
        """
        return await self._request(
            "GET", f"{PATH_SCREENSHOTS}/{core}/{filename}", parse_json=False
        )

    async def async_get_music_status(self) -> dict:
        """Return the current music-player status dict."""
        return await self._request("GET", PATH_MUSIC_STATUS) or {}

    async def async_music_play(self) -> None:
        """Start or resume music playback."""
        await self._request("POST", PATH_MUSIC_PLAY)

    async def async_music_stop(self) -> None:
        """Stop music playback."""
        await self._request("POST", PATH_MUSIC_STOP)

    async def async_music_next(self) -> None:
        """Skip to the next track."""
        await self._request("POST", PATH_MUSIC_NEXT)

    # --- Wallpapers ---
    async def async_get_wallpapers(self) -> dict:
        """Return available wallpapers and the currently active one."""
        return await self._request("GET", PATH_WALLPAPERS) or {}

    async def async_set_wallpaper(self, filename: str) -> None:
        """Set the active wallpaper by filename."""
        await self._request("POST", f"{PATH_WALLPAPERS}/{filename}")

    async def async_clear_wallpaper(self) -> None:
        """Remove the active wallpaper (revert to default)."""
        await self._request("DELETE", PATH_WALLPAPERS)

    # --- INI files ---
    async def async_get_inis(self) -> dict:
        """Return a list of MiSTer.ini profiles and the active index."""
        return await self._request("GET", PATH_INIS) or {}

    async def async_get_ini_values(self, ini_id: int) -> dict:
        """Return all key/value pairs for INI profile *ini_id*."""
        return await self._request("GET", f"{PATH_INIS}/{ini_id}") or {}

    async def async_set_active_ini(self, ini_id: int) -> None:
        """Switch the active MiSTer.ini profile to *ini_id*."""
        await self._request("PUT", PATH_INIS, payload={"ini": ini_id})

    async def async_set_ini_values(self, ini_id: int, values: dict) -> None:
        """Update key/value pairs in INI profile *ini_id*."""
        await self._request("PUT", f"{PATH_INIS}/{ini_id}", payload=values)

    async def async_set_background_mode(self, mode: int) -> None:
        """Set the core/menu background display mode (0 = off, 1 = wallpaper, …)."""
        await self._request("PUT", PATH_CORE_MENU, payload={"mode": mode})

    # --- Music (extended) ---
    async def async_get_music_playlists(self) -> list[str]:
        """Return the list of available music playlist names."""
        return await self._request("GET", PATH_MUSIC_PLAYLIST) or []

    async def async_set_music_playlist(self, name: str) -> None:
        """Activate the music playlist identified by *name*."""
        await self._request("POST", f"{PATH_MUSIC_PLAYLIST}/{name}")

    async def async_set_music_playback(self, mode: str) -> None:
        """Set playback mode (e.g. ``"random"``, ``"loop"``)."""
        await self._request("POST", f"{PATH_MUSIC_PLAYBACK}/{mode}")

    # --- Scripts ---
    async def async_get_scripts(self) -> dict:
        """Return the list of user scripts available on the MiSTer."""
        return await self._request("GET", PATH_SCRIPTS_LIST) or {}

    async def async_launch_script(self, filename: str) -> None:
        """Launch a user script by its filename."""
        await self._request("POST", f"{PATH_SCRIPTS_LAUNCH}/{filename}")

    async def async_open_console(self) -> None:
        """Open the MiSTer console (script terminal)."""
        await self._request("POST", PATH_SCRIPTS_CONSOLE)

    async def async_kill_script(self) -> None:
        """Kill the currently running user script."""
        await self._request("POST", PATH_SCRIPTS_KILL)

    # --- Peers ---
    async def async_get_peers(self) -> list[dict]:
        """Return the list of mrext Remote peers on the local network."""
        data = await self._request("GET", PATH_PEERS) or {}
        return data.get("peers", [])

    # --- Launchers ---
    async def async_launch_path(self, path: str) -> None:
        """Launch a file (ROM, core, script) by its absolute SD-card path."""
        await self._request("POST", PATH_LAUNCH, payload={"path": path})

    async def async_launch_token(self, data: str) -> None:
        """Launch using a short-link token (mrext ``/l/<data>`` endpoint)."""
        await self._request("GET", f"{PATH_LAUNCH_TOKEN}/{data}")

    async def async_create_shortcut(
        self, game_path: str, folder: str, name: str
    ) -> dict:
        """Create a game shortcut in the Favourites/Recents folder.

        Args:
            game_path: Absolute path to the game ROM on the SD card.
            folder: Target folder name (e.g. ``"_Favorites"``).
            name: Display name for the shortcut.

        Returns:
            A dict describing the created shortcut entry.
        """
        return await self._request(
            "POST",
            PATH_LAUNCH_NEW,
            payload={"gamePath": game_path, "folder": folder, "name": name},
        ) or {}

    # --- Raw keyboard ---
    async def async_send_keyboard_raw(self, code: int) -> None:
        """Send a raw HID keyboard scan-code integer."""
        await self._request("POST", f"{PATH_KEYBOARD_RAW}/{code}")

async_close async

async_close() -> None

Close the underlying HTTP session if this client created it.

Source code in src/mister_fpga/client.py

async def async_close(self) -> None:
    """Close the underlying HTTP session if this client created it."""
    if self._owns_session and self._session is not None:
        await self._session.close()
        self._session = None

async_get_status async

async_get_status() -> MisterStatus

Fetch a combined sysinfo + playing snapshot.

Returns:

Name	Type	Description
A	MisterStatus	class:MisterStatus with online=True on success.

Raises:

Type	Description
MisterConnectionError	If either underlying request fails.

Source code in src/mister_fpga/client.py

async def async_get_status(self) -> MisterStatus:
    """Fetch a combined sysinfo + playing snapshot.

    Returns:
        A :class:`MisterStatus` with ``online=True`` on success.

    Raises:
        MisterConnectionError: If either underlying request fails.
    """
    sysinfo = await self._request("GET", PATH_SYSINFO) or {}
    playing = await self._request("GET", PATH_PLAYING) or {}
    ips = sysinfo.get("ips") or []
    disks = sysinfo.get("disks") or []
    disk = disks[0] if disks else {}
    return MisterStatus(
        online=True,
        core=playing.get("core") or None,
        system=playing.get("system") or None,
        system_name=playing.get("systemName") or None,
        game=playing.get("game") or None,
        game_name=playing.get("gameName") or None,
        hostname=sysinfo.get("hostname"),
        version=sysinfo.get("version"),
        ips=ips,
        ip=ips[0] if ips else None,
        updated=sysinfo.get("updated"),
        dns=sysinfo.get("dns"),
        disk_total=disk.get("total"),
        disk_used=disk.get("used"),
        disk_free=disk.get("free"),
    )

async_get_systems async

async_get_systems() -> list[dict]

Return the list of systems known to mrext.

Source code in src/mister_fpga/client.py

async def async_get_systems(self) -> list[dict]:
    """Return the list of systems known to mrext."""
    return await self._request("GET", PATH_SYSTEMS) or []

async_launch_system async

async_launch_system(system_id: str) -> None

Launch the default game/core for system_id.

Source code in src/mister_fpga/client.py

async def async_launch_system(self, system_id: str) -> None:
    """Launch the default game/core for *system_id*."""
    await self._request("POST", f"{PATH_SYSTEMS}/{system_id}")

async_launch_game async

async_launch_game(path: str) -> None

Launch a game by its absolute path on the MiSTer SD card.

Source code in src/mister_fpga/client.py

async def async_launch_game(self, path: str) -> None:
    """Launch a game by its absolute path on the MiSTer SD card."""
    await self._request("POST", PATH_GAMES_LAUNCH, payload={"path": path})

async_launch_menu async

async_launch_menu() -> None

Return to the MiSTer main menu.

Source code in src/mister_fpga/client.py

async def async_launch_menu(self) -> None:
    """Return to the MiSTer main menu."""
    await self._request("POST", PATH_LAUNCH_MENU)

async_search_games async

async_search_games(query: str, system: str = 'all') -> dict

Search the game index.

Parameters:

Name	Type	Description	Default
query	str	Free-text search string.	required
system	str	Limit results to a system slug, or "all" (default).	'all'

Returns:

Type	Description
dict	A dict with a results key containing matching game entries.

Source code in src/mister_fpga/client.py

async def async_search_games(self, query: str, system: str = "all") -> dict:
    """Search the game index.

    Args:
        query: Free-text search string.
        system: Limit results to a system slug, or ``"all"`` (default).

    Returns:
        A dict with a ``results`` key containing matching game entries.
    """
    return (
        await self._request(
            "POST", PATH_GAMES_SEARCH, payload={"data": query, "system": system}
        )
        or {}
    )

async_index_games async

async_index_games() -> None

Trigger a background re-index of the game library.

Source code in src/mister_fpga/client.py

async def async_index_games(self) -> None:
    """Trigger a background re-index of the game library."""
    await self._request("POST", PATH_GAMES_INDEX)

async_send_keyboard async

async_send_keyboard(name: str) -> None

Send a named virtual-keyboard event (see KEYBOARD_NAMES).

Source code in src/mister_fpga/client.py

async def async_send_keyboard(self, name: str) -> None:
    """Send a named virtual-keyboard event (see ``KEYBOARD_NAMES``)."""
    await self._request("POST", f"{PATH_KEYBOARD}/{name}")

async_reboot async

async_reboot() -> None

Reboot the MiSTer device.

Source code in src/mister_fpga/client.py

async def async_reboot(self) -> None:
    """Reboot the MiSTer device."""
    await self._request("POST", PATH_REBOOT)

async_restart_remote async

async_restart_remote() -> None

Restart the mrext Remote service on the MiSTer.

Source code in src/mister_fpga/client.py

async def async_restart_remote(self) -> None:
    """Restart the mrext Remote service on the MiSTer."""
    await self._request("POST", PATH_RESTART_REMOTE)

async_take_screenshot async

async_take_screenshot() -> None

Capture a screenshot on the MiSTer and save it server-side.

Source code in src/mister_fpga/client.py

async def async_take_screenshot(self) -> None:
    """Capture a screenshot on the MiSTer and save it server-side."""
    await self._request("POST", PATH_SCREENSHOTS)

async_get_screenshots async

async_get_screenshots() -> list[dict]

Return metadata for all saved screenshots.

Source code in src/mister_fpga/client.py

async def async_get_screenshots(self) -> list[dict]:
    """Return metadata for all saved screenshots."""
    return await self._request("GET", PATH_SCREENSHOTS) or []

async_get_screenshot_image async

async_get_screenshot_image(core: str, filename: str) -> bytes

Download a screenshot image as raw bytes.

Parameters:

Name	Type	Description	Default
core	str	Core name sub-directory (e.g. "SNES").	required
filename	str	Screenshot filename within that sub-directory.	required

Returns:

Type	Description
bytes	Raw image bytes (typically PNG).

Raises:

Type	Description
MisterConnectionError	If the image cannot be retrieved.

Source code in src/mister_fpga/client.py

async def async_get_screenshot_image(self, core: str, filename: str) -> bytes:
    """Download a screenshot image as raw bytes.

    Args:
        core: Core name sub-directory (e.g. ``"SNES"``).
        filename: Screenshot filename within that sub-directory.

    Returns:
        Raw image bytes (typically PNG).

    Raises:
        MisterConnectionError: If the image cannot be retrieved.
    """
    return await self._request(
        "GET", f"{PATH_SCREENSHOTS}/{core}/{filename}", parse_json=False
    )

async_get_music_status async

async_get_music_status() -> dict

Return the current music-player status dict.

Source code in src/mister_fpga/client.py

async def async_get_music_status(self) -> dict:
    """Return the current music-player status dict."""
    return await self._request("GET", PATH_MUSIC_STATUS) or {}

async_music_play async

async_music_play() -> None

Start or resume music playback.

Source code in src/mister_fpga/client.py

async def async_music_play(self) -> None:
    """Start or resume music playback."""
    await self._request("POST", PATH_MUSIC_PLAY)

async_music_stop async

async_music_stop() -> None

Stop music playback.

Source code in src/mister_fpga/client.py

async def async_music_stop(self) -> None:
    """Stop music playback."""
    await self._request("POST", PATH_MUSIC_STOP)

async_music_next async

async_music_next() -> None

Skip to the next track.

Source code in src/mister_fpga/client.py

async def async_music_next(self) -> None:
    """Skip to the next track."""
    await self._request("POST", PATH_MUSIC_NEXT)

async_get_wallpapers async

async_get_wallpapers() -> dict

Return available wallpapers and the currently active one.

Source code in src/mister_fpga/client.py

async def async_get_wallpapers(self) -> dict:
    """Return available wallpapers and the currently active one."""
    return await self._request("GET", PATH_WALLPAPERS) or {}

async_set_wallpaper async

async_set_wallpaper(filename: str) -> None

Set the active wallpaper by filename.

Source code in src/mister_fpga/client.py

async def async_set_wallpaper(self, filename: str) -> None:
    """Set the active wallpaper by filename."""
    await self._request("POST", f"{PATH_WALLPAPERS}/{filename}")

async_clear_wallpaper async

async_clear_wallpaper() -> None

Remove the active wallpaper (revert to default).

Source code in src/mister_fpga/client.py

async def async_clear_wallpaper(self) -> None:
    """Remove the active wallpaper (revert to default)."""
    await self._request("DELETE", PATH_WALLPAPERS)

async_get_inis async

async_get_inis() -> dict

Return a list of MiSTer.ini profiles and the active index.

Source code in src/mister_fpga/client.py

async def async_get_inis(self) -> dict:
    """Return a list of MiSTer.ini profiles and the active index."""
    return await self._request("GET", PATH_INIS) or {}

async_get_ini_values async

async_get_ini_values(ini_id: int) -> dict

Return all key/value pairs for INI profile ini_id.

Source code in src/mister_fpga/client.py

async def async_get_ini_values(self, ini_id: int) -> dict:
    """Return all key/value pairs for INI profile *ini_id*."""
    return await self._request("GET", f"{PATH_INIS}/{ini_id}") or {}

async_set_active_ini async

async_set_active_ini(ini_id: int) -> None

Switch the active MiSTer.ini profile to ini_id.

Source code in src/mister_fpga/client.py

async def async_set_active_ini(self, ini_id: int) -> None:
    """Switch the active MiSTer.ini profile to *ini_id*."""
    await self._request("PUT", PATH_INIS, payload={"ini": ini_id})

async_set_ini_values async

async_set_ini_values(ini_id: int, values: dict) -> None

Update key/value pairs in INI profile ini_id.

Source code in src/mister_fpga/client.py

async def async_set_ini_values(self, ini_id: int, values: dict) -> None:
    """Update key/value pairs in INI profile *ini_id*."""
    await self._request("PUT", f"{PATH_INIS}/{ini_id}", payload=values)

async_set_background_mode async

async_set_background_mode(mode: int) -> None

Set the core/menu background display mode (0 = off, 1 = wallpaper, …).

Source code in src/mister_fpga/client.py

async def async_set_background_mode(self, mode: int) -> None:
    """Set the core/menu background display mode (0 = off, 1 = wallpaper, …)."""
    await self._request("PUT", PATH_CORE_MENU, payload={"mode": mode})

async_get_music_playlists async

async_get_music_playlists() -> list[str]

Return the list of available music playlist names.

Source code in src/mister_fpga/client.py

async def async_get_music_playlists(self) -> list[str]:
    """Return the list of available music playlist names."""
    return await self._request("GET", PATH_MUSIC_PLAYLIST) or []

async_set_music_playlist async

async_set_music_playlist(name: str) -> None

Activate the music playlist identified by name.

Source code in src/mister_fpga/client.py

async def async_set_music_playlist(self, name: str) -> None:
    """Activate the music playlist identified by *name*."""
    await self._request("POST", f"{PATH_MUSIC_PLAYLIST}/{name}")

async_set_music_playback async

async_set_music_playback(mode: str) -> None

Set playback mode (e.g. "random", "loop").

Source code in src/mister_fpga/client.py

async def async_set_music_playback(self, mode: str) -> None:
    """Set playback mode (e.g. ``"random"``, ``"loop"``)."""
    await self._request("POST", f"{PATH_MUSIC_PLAYBACK}/{mode}")

async_get_scripts async

async_get_scripts() -> dict

Return the list of user scripts available on the MiSTer.

Source code in src/mister_fpga/client.py

async def async_get_scripts(self) -> dict:
    """Return the list of user scripts available on the MiSTer."""
    return await self._request("GET", PATH_SCRIPTS_LIST) or {}

async_launch_script async

async_launch_script(filename: str) -> None

Launch a user script by its filename.

Source code in src/mister_fpga/client.py

async def async_launch_script(self, filename: str) -> None:
    """Launch a user script by its filename."""
    await self._request("POST", f"{PATH_SCRIPTS_LAUNCH}/{filename}")

async_open_console async

async_open_console() -> None

Open the MiSTer console (script terminal).

Source code in src/mister_fpga/client.py

async def async_open_console(self) -> None:
    """Open the MiSTer console (script terminal)."""
    await self._request("POST", PATH_SCRIPTS_CONSOLE)

async_kill_script async

async_kill_script() -> None

Kill the currently running user script.

Source code in src/mister_fpga/client.py

async def async_kill_script(self) -> None:
    """Kill the currently running user script."""
    await self._request("POST", PATH_SCRIPTS_KILL)

async_get_peers async

async_get_peers() -> list[dict]

Return the list of mrext Remote peers on the local network.

Source code in src/mister_fpga/client.py

async def async_get_peers(self) -> list[dict]:
    """Return the list of mrext Remote peers on the local network."""
    data = await self._request("GET", PATH_PEERS) or {}
    return data.get("peers", [])

async_launch_path async

async_launch_path(path: str) -> None

Launch a file (ROM, core, script) by its absolute SD-card path.

Source code in src/mister_fpga/client.py

async def async_launch_path(self, path: str) -> None:
    """Launch a file (ROM, core, script) by its absolute SD-card path."""
    await self._request("POST", PATH_LAUNCH, payload={"path": path})

async_launch_token async

async_launch_token(data: str) -> None

Launch using a short-link token (mrext /l/<data> endpoint).

Source code in src/mister_fpga/client.py

async def async_launch_token(self, data: str) -> None:
    """Launch using a short-link token (mrext ``/l/<data>`` endpoint)."""
    await self._request("GET", f"{PATH_LAUNCH_TOKEN}/{data}")

async_create_shortcut async

async_create_shortcut(game_path: str, folder: str, name: str) -> dict

Create a game shortcut in the Favourites/Recents folder.

Parameters:

Name	Type	Description	Default
game_path	str	Absolute path to the game ROM on the SD card.	required
folder	str	Target folder name (e.g. "_Favorites").	required
name	str	Display name for the shortcut.	required

Returns:

Type	Description
dict	A dict describing the created shortcut entry.

Source code in src/mister_fpga/client.py

async def async_create_shortcut(
    self, game_path: str, folder: str, name: str
) -> dict:
    """Create a game shortcut in the Favourites/Recents folder.

    Args:
        game_path: Absolute path to the game ROM on the SD card.
        folder: Target folder name (e.g. ``"_Favorites"``).
        name: Display name for the shortcut.

    Returns:
        A dict describing the created shortcut entry.
    """
    return await self._request(
        "POST",
        PATH_LAUNCH_NEW,
        payload={"gamePath": game_path, "folder": folder, "name": name},
    ) or {}

async_send_keyboard_raw async

async_send_keyboard_raw(code: int) -> None

Send a raw HID keyboard scan-code integer.

Source code in src/mister_fpga/client.py

async def async_send_keyboard_raw(self, code: int) -> None:
    """Send a raw HID keyboard scan-code integer."""
    await self._request("POST", f"{PATH_KEYBOARD_RAW}/{code}")

mister_fpga.MisterStatus dataclass

Snapshot of the MiSTer device state at a point in time.

Returned by :meth:MisterClient.async_get_status and mutated in-place by :func:~mister_fpga.apply_ws_message.

Attributes:

Name	Type	Description
online	bool	True when the API responded successfully.
core	str | None	Short core identifier (e.g. "SNES"), or None when the menu is active.
system	str | None	System slug used by mrext (e.g. "snes").
system_name	str | None	Human-readable system name (e.g. "Super Nintendo").
game	str | None	Full path of the running game ROM, or None.
game_name	str | None	Basename of the game without extension, or None.
hostname	str | None	MiSTer hostname as reported by the Remote API.
version	str | None	mrext Remote version string.
ip	str | None	Primary IP address of the MiSTer device.
ips	list[str]	All IP addresses reported by the device.
updated	str | None	ISO-8601 timestamp of the last sysinfo update.
dns	str | None	DNS server address reported by the device.
disk_total	int | None	Total SD-card space in bytes (first disk).
disk_used	int | None	Used SD-card space in bytes (first disk).
disk_free	int | None	Free SD-card space in bytes (first disk).

Source code in src/mister_fpga/client.py

@dataclass
class MisterStatus:
    """Snapshot of the MiSTer device state at a point in time.

    Returned by :meth:`MisterClient.async_get_status` and mutated in-place by
    :func:`~mister_fpga.apply_ws_message`.

    Attributes:
        online: ``True`` when the API responded successfully.
        core: Short core identifier (e.g. ``"SNES"``), or ``None`` when the
            menu is active.
        system: System slug used by mrext (e.g. ``"snes"``).
        system_name: Human-readable system name (e.g. ``"Super Nintendo"``).
        game: Full path of the running game ROM, or ``None``.
        game_name: Basename of the game without extension, or ``None``.
        hostname: MiSTer hostname as reported by the Remote API.
        version: mrext Remote version string.
        ip: Primary IP address of the MiSTer device.
        ips: All IP addresses reported by the device.
        updated: ISO-8601 timestamp of the last sysinfo update.
        dns: DNS server address reported by the device.
        disk_total: Total SD-card space in bytes (first disk).
        disk_used: Used SD-card space in bytes (first disk).
        disk_free: Free SD-card space in bytes (first disk).
    """

    online: bool = False
    core: str | None = None
    system: str | None = None
    system_name: str | None = None
    game: str | None = None
    game_name: str | None = None
    hostname: str | None = None
    version: str | None = None
    ip: str | None = None
    ips: list[str] = field(default_factory=list)
    updated: str | None = None
    dns: str | None = None
    disk_total: int | None = None
    disk_used: int | None = None
    disk_free: int | None = None

    @property
    def is_running_game(self) -> bool:
        """True when a real core/game (not the menu) is running."""
        if not self.online:
            return False
        core = (self.core or "").strip().lower()
        return bool(core) and core not in ("menu", "none")

is_running_game property

is_running_game: bool

True when a real core/game (not the menu) is running.

mister_fpga.MisterWebSocketClient

Reconnecting WebSocket client for real-time MiSTer Remote events.

Connects to the mrext Remote WebSocket endpoint and invokes a callback for every TEXT frame received. If the connection drops the client waits reconnect_delay seconds and reconnects automatically.

Parameters:

Name	Type	Description	Default
host	str	IP address or hostname of the MiSTer device.	required
port	int	mrext Remote port (default 8182).	8182
session	ClientSession | None	Optional shared aiohttp.ClientSession. When None the client creates and owns its own session.	None
reconnect_delay	int	Seconds to wait before reconnecting after a connection loss (default 5).	5

Source code in src/mister_fpga/websocket.py

class MisterWebSocketClient:
    """Reconnecting WebSocket client for real-time MiSTer Remote events.

    Connects to the mrext Remote WebSocket endpoint and invokes a callback for
    every TEXT frame received.  If the connection drops the client waits
    *reconnect_delay* seconds and reconnects automatically.

    Args:
        host: IP address or hostname of the MiSTer device.
        port: mrext Remote port (default ``8182``).
        session: Optional shared ``aiohttp.ClientSession``.  When ``None`` the
            client creates and owns its own session.
        reconnect_delay: Seconds to wait before reconnecting after a
            connection loss (default ``5``).
    """

    def __init__(
        self,
        host: str,
        port: int = 8182,
        *,
        session: aiohttp.ClientSession | None = None,
        reconnect_delay: int = 5,
    ) -> None:
        self.host = host
        self.port = port
        self._session = session
        self._owns_session = session is None
        self._reconnect_delay = reconnect_delay
        self._stop = False

    @property
    def url(self) -> str:
        """Full WebSocket URL derived from *host* and *port*."""
        return f"ws://{self.host}:{self.port}{WS_PATH}"

    async def listen(
        self,
        on_message: Callable[[str], None | Awaitable[None]],
    ) -> None:
        """Run the reconnect loop, calling *on_message* for each TEXT frame.

        Blocks until :meth:`stop` is called or the task is cancelled.
        *on_message* may be a plain function or a coroutine function.

        Args:
            on_message: Callable ``(text: str) -> None | Awaitable``.
        """
        owns = self._session is None
        session = self._session or aiohttp.ClientSession()
        try:
            while not self._stop:
                try:
                    async with session.ws_connect(self.url, heartbeat=30) as ws:
                        async for msg in ws:
                            if msg.type == aiohttp.WSMsgType.TEXT:
                                result = on_message(msg.data)
                                if hasattr(result, "__await__"):
                                    await result
                            elif msg.type in (
                                aiohttp.WSMsgType.CLOSED,
                                aiohttp.WSMsgType.ERROR,
                            ):
                                break
                except (aiohttp.ClientError, TimeoutError):
                    pass
                if self._stop:
                    break
                await asyncio.sleep(self._reconnect_delay)
        finally:
            if owns:
                await session.close()

    def stop(self) -> None:
        """Signal the :meth:`listen` loop to exit after the current iteration."""
        self._stop = True

url property

url: str

Full WebSocket URL derived from host and port.

listen async

listen(on_message: Callable[[str], None | Awaitable[None]]) -> None

Run the reconnect loop, calling on_message for each TEXT frame.

Blocks until :meth:stop is called or the task is cancelled. on_message may be a plain function or a coroutine function.

Parameters:

Name	Type	Description	Default
on_message	Callable[[str], None | Awaitable[None]]	Callable (text: str) -> None | Awaitable.	required

Source code in src/mister_fpga/websocket.py

async def listen(
    self,
    on_message: Callable[[str], None | Awaitable[None]],
) -> None:
    """Run the reconnect loop, calling *on_message* for each TEXT frame.

    Blocks until :meth:`stop` is called or the task is cancelled.
    *on_message* may be a plain function or a coroutine function.

    Args:
        on_message: Callable ``(text: str) -> None | Awaitable``.
    """
    owns = self._session is None
    session = self._session or aiohttp.ClientSession()
    try:
        while not self._stop:
            try:
                async with session.ws_connect(self.url, heartbeat=30) as ws:
                    async for msg in ws:
                        if msg.type == aiohttp.WSMsgType.TEXT:
                            result = on_message(msg.data)
                            if hasattr(result, "__await__"):
                                await result
                        elif msg.type in (
                            aiohttp.WSMsgType.CLOSED,
                            aiohttp.WSMsgType.ERROR,
                        ):
                            break
            except (aiohttp.ClientError, TimeoutError):
                pass
            if self._stop:
                break
            await asyncio.sleep(self._reconnect_delay)
    finally:
        if owns:
            await session.close()

stop

stop() -> None

Signal the :meth:listen loop to exit after the current iteration.

Source code in src/mister_fpga/websocket.py

def stop(self) -> None:
    """Signal the :meth:`listen` loop to exit after the current iteration."""
    self._stop = True

mister_fpga.apply_ws_message

apply_ws_message(message: str, status: MisterStatus, menu_path: str | None, index_state: tuple[bool, bool]) -> tuple[MisterStatus, str | None, tuple[bool, bool]]

Apply one WebSocket text frame to the current state triple.

This is a pure reducer — it never mutates its arguments. Pass it every raw text frame received from :class:MisterWebSocketClient to keep a local :class:~mister_fpga.MisterStatus in sync without polling the REST API.

Understood prefixes (prefix:rest):

• coreRunning — updates :attr:~mister_fpga.MisterStatus.core.
• gameRunning — updates :attr:~mister_fpga.MisterStatus.game and :attr:~mister_fpga.MisterStatus.game_name.
• menuNavigation — updates menu_path.
• indexStatus — updates index_state (exists, in_progress).

Parameters:

Name	Type	Description	Default
message	str	Raw text frame received from the WebSocket.	required
status	MisterStatus	Current device-state snapshot.	required
menu_path	str | None	Current menu navigation path, or None.	required
index_state	tuple[bool, bool]	Tuple (index_exists, index_in_progress).	required

Returns:

Type	Description
MisterStatus	A new (status, menu_path, index_state) triple reflecting the
str | None	change encoded in message. Unrecognised prefixes are passed through
tuple[bool, bool]	unchanged.

Source code in src/mister_fpga/websocket.py

def apply_ws_message(
    message: str,
    status: MisterStatus,
    menu_path: str | None,
    index_state: tuple[bool, bool],
) -> tuple[MisterStatus, str | None, tuple[bool, bool]]:
    """Apply one WebSocket text frame to the current state triple.

    This is a *pure* reducer — it never mutates its arguments.  Pass it every
    raw text frame received from :class:`MisterWebSocketClient` to keep a
    local :class:`~mister_fpga.MisterStatus` in sync without polling the REST
    API.

    Understood prefixes (``prefix:rest``):

    * ``coreRunning`` — updates :attr:`~mister_fpga.MisterStatus.core`.
    * ``gameRunning`` — updates :attr:`~mister_fpga.MisterStatus.game` and
      :attr:`~mister_fpga.MisterStatus.game_name`.
    * ``menuNavigation`` — updates *menu_path*.
    * ``indexStatus`` — updates *index_state* ``(exists, in_progress)``.

    Args:
        message: Raw text frame received from the WebSocket.
        status: Current device-state snapshot.
        menu_path: Current menu navigation path, or ``None``.
        index_state: Tuple ``(index_exists, index_in_progress)``.

    Returns:
        A new ``(status, menu_path, index_state)`` triple reflecting the
        change encoded in *message*.  Unrecognised prefixes are passed through
        unchanged.
    """
    prefix, _, rest = message.partition(":")
    if prefix == "coreRunning":
        core = rest.strip() or None
        if core is None:
            return (
                replace(status, core=None, game=None, game_name=None),
                menu_path,
                index_state,
            )
        return replace(status, core=core), menu_path, index_state
    if prefix == "gameRunning":
        rest = rest.strip()
        if not rest:
            return replace(status, game=None, game_name=None), menu_path, index_state
        _, _, name = rest.partition("/")
        game_name = name.rsplit(".", 1)[0] if name else None
        return replace(status, game=rest, game_name=game_name), menu_path, index_state
    if prefix == "menuNavigation":
        return status, rest.strip() or None, index_state
    if prefix == "indexStatus":
        parts = rest.split(",")
        exists = len(parts) > 0 and parts[0] == "y"
        in_progress = len(parts) > 1 and parts[1] == "y"
        return status, menu_path, (exists, in_progress)
    return status, menu_path, index_state

mister_fpga.MisterSSH

Persistent asyncssh connection for MiSTer telemetry probes.

Maintains a single SSH connection that is re-established transparently on failure. SSH telemetry is strictly best-effort: :meth:async_probe always returns a dict (possibly empty) and never raises.

Parameters:

Name	Type	Description	Default
host	str	IP address or hostname of the MiSTer device.	required
port	int	SSH port (default 22).	required
username	str	SSH username (default "root").	required
password	str	SSH password (often "1" on stock MiSTer).	required

Source code in src/mister_fpga/ssh.py

class MisterSSH:
    """Persistent asyncssh connection for MiSTer telemetry probes.

    Maintains a single SSH connection that is re-established transparently on
    failure.  SSH telemetry is strictly best-effort: :meth:`async_probe`
    always returns a dict (possibly empty) and never raises.

    Args:
        host: IP address or hostname of the MiSTer device.
        port: SSH port (default ``22``).
        username: SSH username (default ``"root"``).
        password: SSH password (often ``"1"`` on stock MiSTer).
    """

    def __init__(self, host: str, port: int, username: str, password: str) -> None:
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self._conn = None

    async def _ensure(self) -> None:
        if self._conn is not None:
            return
        import asyncssh

        self._conn = await asyncssh.connect(
            self.host,
            port=self.port,
            username=self.username,
            password=self.password,
            known_hosts=None,
        )

    async def async_probe(self) -> dict:
        """Run the SSH probe command and return parsed telemetry.

        Returns:
            Telemetry dict as returned by :func:`parse_ssh_probe`, or ``{}``
            if the SSH connection or command fails.
        """
        try:
            await self._ensure()
            result = await self._conn.run(SSH_PROBE_CMD, check=False, timeout=10)
            return parse_ssh_probe(result.stdout or "")
        except Exception as err:  # noqa: BLE001 - asyncssh raises many types; SSH is best-effort and must never break the HTTP integration
            _LOGGER.debug("MiSTer SSH probe failed: %s", err)
            self._conn = None
            return {}

    async def async_close(self) -> None:
        """Close the underlying SSH connection if open."""
        if self._conn is not None:
            self._conn.close()
            self._conn = None

async_probe async

async_probe() -> dict

Run the SSH probe command and return parsed telemetry.

Returns:

Type	Description
dict	Telemetry dict as returned by :func:parse_ssh_probe, or {}
dict	if the SSH connection or command fails.

Source code in src/mister_fpga/ssh.py

async def async_probe(self) -> dict:
    """Run the SSH probe command and return parsed telemetry.

    Returns:
        Telemetry dict as returned by :func:`parse_ssh_probe`, or ``{}``
        if the SSH connection or command fails.
    """
    try:
        await self._ensure()
        result = await self._conn.run(SSH_PROBE_CMD, check=False, timeout=10)
        return parse_ssh_probe(result.stdout or "")
    except Exception as err:  # noqa: BLE001 - asyncssh raises many types; SSH is best-effort and must never break the HTTP integration
        _LOGGER.debug("MiSTer SSH probe failed: %s", err)
        self._conn = None
        return {}

async_close async

async_close() -> None

Close the underlying SSH connection if open.

Source code in src/mister_fpga/ssh.py

async def async_close(self) -> None:
    """Close the underlying SSH connection if open."""
    if self._conn is not None:
        self._conn.close()
        self._conn = None

mister_fpga.parse_ssh_probe

parse_ssh_probe(raw: str) -> dict

Parse the batched SSH probe output into a telemetry dict.

The probe command (SSH_PROBE_CMD) concatenates five fields separated by |||: active core name, uptime, load average, memory info, and firmware timestamp. Any missing or malformed field is silently coerced to None.

Parameters:

Name	Type	Description	Default
raw	str	Raw stdout string returned by running SSH_PROBE_CMD on the MiSTer over SSH.	required

Returns:

Type	Description
dict	A dict with the following keys:
dict	active_core (str | None) — name from /tmp/CORENAME.
dict	uptime_seconds (int | None) — seconds since last boot.
dict	cpu_load_1m (float | None) — 1-minute load average.
dict	memory_used_percent (float | None) — RAM usage 0–100.
dict	firmware_timestamp (int | None) — mtime of the MiSTer binary.

Source code in src/mister_fpga/ssh.py

def parse_ssh_probe(raw: str) -> dict:
    """Parse the batched SSH probe output into a telemetry dict.

    The probe command (``SSH_PROBE_CMD``) concatenates five fields separated by
    ``|||``: active core name, uptime, load average, memory info, and firmware
    timestamp.  Any missing or malformed field is silently coerced to ``None``.

    Args:
        raw: Raw stdout string returned by running ``SSH_PROBE_CMD`` on the
            MiSTer over SSH.

    Returns:
        A dict with the following keys:

        * ``active_core`` (``str | None``) — name from ``/tmp/CORENAME``.
        * ``uptime_seconds`` (``int | None``) — seconds since last boot.
        * ``cpu_load_1m`` (``float | None``) — 1-minute load average.
        * ``memory_used_percent`` (``float | None``) — RAM usage 0–100.
        * ``firmware_timestamp`` (``int | None``) — mtime of the MiSTer binary.
    """
    parts = [p.strip() for p in raw.split(_SEP)]
    while len(parts) < 5:
        parts.append("")
    core, uptime_s, load_s, mem_s, fw_s = parts[:5]

    def _int(value: str) -> int | None:
        try:
            return int(float(value))
        except (TypeError, ValueError):
            return None

    uptime = _int(uptime_s.split()[0]) if uptime_s else None
    load_1m = None
    if load_s:
        try:
            load_1m = float(load_s.split()[0])
        except (ValueError, IndexError):
            load_1m = None
    mem_used_pct = None
    mem_lines = [m for m in mem_s.splitlines() if m]
    if len(mem_lines) >= 2:
        total = _int(mem_lines[0])
        avail = _int(mem_lines[1])
        if total:
            mem_used_pct = round((total - avail) / total * 100, 1)
    fw_ts = _int(fw_s) if fw_s else None

    return {
        "active_core": core or None,
        "uptime_seconds": uptime,
        "cpu_load_1m": load_1m,
        "memory_used_percent": mem_used_pct,
        "firmware_timestamp": fw_ts,
    }

mister_fpga.MisterConnectionError

Bases: Exception

Raised when the MiSTer Remote API is unreachable or returns an error.

This exception wraps aiohttp.ClientError, TimeoutError, and invalid-JSON responses so callers have a single exception type to catch.

Source code in src/mister_fpga/client.py

class MisterConnectionError(Exception):
    """Raised when the MiSTer Remote API is unreachable or returns an error.

    This exception wraps ``aiohttp.ClientError``, ``TimeoutError``, and
    invalid-JSON responses so callers have a single exception type to catch.
    """