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()
                if resp.content_type == "text/html":
                    raise MisterUnsupportedError(
                        f"{method} {url} is not an API endpoint on this Remote"
                    )
                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 = next(
            (d for d in disks if d.get("path") == "/media/fat"),
            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"),
            disks=disks,
            core_session_time=playing.get("coreSessionTime"),
            game_session_time=playing.get("gameSessionTime"),
        )

    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", page: int | None = None
    ) -> dict:
        """Search the game index.

        Args:
            query: Free-text search string.
            system: Limit results to a system id, or ``"all"`` (default).
            page: 1-based result page (Remote >= 0.5 paginates; 0.4 ignores it
                and caps results at 500).

        Returns:
            ``{"data": [{"system": {...}, "name", "path"}], "total",
            "pageSize", "page"}``.
        """
        payload: dict[str, Any] = {"query": query, "system": system}
        if page is not None:
            payload["page"] = page
        return await self._request("POST", PATH_GAMES_SEARCH, payload=payload) or {}

    async def async_get_search_systems(self) -> list[dict]:
        """Return the systems present in the game index (``[{id, name}]``)."""
        data = await self._request("GET", PATH_SEARCH_SYSTEMS) or {}
        return data.get("systems", []) if isinstance(data, dict) else []

    async def async_browse_games(self, path: str | None = None) -> dict:
        """List a games folder (``{"up", "items"}``); root when *path* is None."""
        payload = {"path": _safe_relative(path)} if path else {}
        return await self._request("POST", PATH_GAMES_VIEW, payload=payload) or {}

    async def async_browse_menu(self, path: str = "") -> dict:
        """List a folder of the MiSTer menu tree (``{"up", "items": [...]}``).

        Raises:
            ValueError: If *path* contains ``..`` (Remote 0.4 does not contain it).
        """
        return (
            await self._request(
                "POST", PATH_MENU_VIEW, payload={"path": _safe_relative(path)}
            )
            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."""
        data = await self._request("GET", PATH_SCREENSHOTS)
        return data if isinstance(data, list) else []

    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}/{_seg(core)}/{_seg(filename)}", parse_json=False
        )

    async def async_delete_screenshot(self, core: str, filename: str) -> None:
        """Delete one saved screenshot."""
        await self._request(
            "DELETE", f"{PATH_SCREENSHOTS}/{_seg(core)}/{_seg(filename)}"
        )

    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}/{_seg(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 menu background mode.

        Args:
            mode: One of :data:`~mister_fpga.const.BACKGROUND_MODES` (0 none,
                2 wallpaper, 4/6 horizontal bars, 8/10 vertical bars,
                12 spectrum, 14 black).

        Raises:
            ValueError: For any other value (the server rejects odd modes).
        """
        if mode not in BACKGROUND_MODES:
            raise ValueError(f"invalid background mode {mode}")
        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}/{_seg(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}/{_seg(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}/{_seg(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 a token: a path, or a ``**command:`` such as ``**random:snes``.

        The text is base64url-encoded (no padding) as the ``/l/<data>``
        endpoint requires; pass it in plain form.
        """
        encoded = base64.urlsafe_b64encode(data.encode()).decode().rstrip("=")
        await self._request("GET", f"{PATH_LAUNCH_TOKEN}/{encoded}")

    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:
        """Tap a Linux evdev ``KEY_*`` code (e.g. 28 = Enter).

        A negative code sends Shift + ``abs(code)``.
        """
        await self._request("POST", f"{PATH_KEYBOARD_RAW}/{code}")

    # --- Remote >= 0.5 ---
    async def async_get_screen(self) -> dict:
        """Return the current framebuffer resolution ``{width, height}``."""
        return await self._request("GET", PATH_SCREEN) or {}

    async def async_get_playlog_status(self) -> dict:
        """PlayLog state: ``{installed, running, database, recents, startup}``."""
        return await self._request("GET", PATH_PLAYLOG_STATUS) or {}

    async def async_get_playlog_summary(
        self, *, top: int | None = None, days: int | None = None
    ) -> dict:
        """Play-time summary: totals (seconds), streaks, top games/systems, per-day."""
        params = "&".join(
            f"{k}={v}" for k, v in (("top", top), ("days", days)) if v is not None
        )
        path = PATH_PLAYLOG_SUMMARY + (f"?{params}" if params else "")
        return await self._request("GET", path) or {}

    async def async_get_playlog_games(
        self, *, sort: str | None = None, limit: int | None = None, offset: int = 0
    ) -> dict:
        """Per-game play stats ``{"games": [...], "total"}``."""
        params = [f"offset={offset}"]
        if sort:
            params.append(f"sort={_seg(sort)}")
        if limit:
            params.append(f"limit={limit}")
        return (
            await self._request("GET", f"{PATH_PLAYLOG_GAMES}?{'&'.join(params)}")
            or {}
        )

    async def async_get_playlog_sessions(
        self, *, limit: int | None = None, offset: int = 0
    ) -> dict:
        """Recent play sessions ``{"sessions": [...], "total"}``."""
        params = f"offset={offset}" + (f"&limit={limit}" if limit else "")
        return await self._request("GET", f"{PATH_PLAYLOG_SESSIONS}?{params}") or {}

    async def async_get_remote_log(self) -> bytes:
        """Download the Remote service log."""
        return await self._request("GET", PATH_REMOTE_LOG, parse_json=False)

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 = next(
        (d for d in disks if d.get("path") == "/media/fat"),
        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"),
        disks=disks,
        core_session_time=playing.get("coreSessionTime"),
        game_session_time=playing.get("gameSessionTime"),
    )

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', page: int | None = None) -> dict

Search the game index.

Parameters:

Name	Type	Description	Default
query	str	Free-text search string.	required
system	str	Limit results to a system id, or "all" (default).	'all'
page	int | None	1-based result page (Remote >= 0.5 paginates; 0.4 ignores it and caps results at 500).	None

Returns:

Type	Description
dict	``{"data": [{"system": {...}, "name", "path"}], "total",
dict	"pageSize", "page"}``.

Source code in src/mister_fpga/client.py

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

    Args:
        query: Free-text search string.
        system: Limit results to a system id, or ``"all"`` (default).
        page: 1-based result page (Remote >= 0.5 paginates; 0.4 ignores it
            and caps results at 500).

    Returns:
        ``{"data": [{"system": {...}, "name", "path"}], "total",
        "pageSize", "page"}``.
    """
    payload: dict[str, Any] = {"query": query, "system": system}
    if page is not None:
        payload["page"] = page
    return await self._request("POST", PATH_GAMES_SEARCH, payload=payload) or {}

async_get_search_systems async

async_get_search_systems() -> list[dict]

Return the systems present in the game index ([{id, name}]).

Source code in src/mister_fpga/client.py

async def async_get_search_systems(self) -> list[dict]:
    """Return the systems present in the game index (``[{id, name}]``)."""
    data = await self._request("GET", PATH_SEARCH_SYSTEMS) or {}
    return data.get("systems", []) if isinstance(data, dict) else []

async_browse_games async

async_browse_games(path: str | None = None) -> dict

List a games folder ({"up", "items"}); root when path is None.

Source code in src/mister_fpga/client.py

async def async_browse_games(self, path: str | None = None) -> dict:
    """List a games folder (``{"up", "items"}``); root when *path* is None."""
    payload = {"path": _safe_relative(path)} if path else {}
    return await self._request("POST", PATH_GAMES_VIEW, payload=payload) or {}

async_browse_menu async

async_browse_menu(path: str = '') -> dict

List a folder of the MiSTer menu tree ({"up", "items": [...]}).

Raises:

Type	Description
ValueError	If path contains .. (Remote 0.4 does not contain it).

Source code in src/mister_fpga/client.py

async def async_browse_menu(self, path: str = "") -> dict:
    """List a folder of the MiSTer menu tree (``{"up", "items": [...]}``).

    Raises:
        ValueError: If *path* contains ``..`` (Remote 0.4 does not contain it).
    """
    return (
        await self._request(
            "POST", PATH_MENU_VIEW, payload={"path": _safe_relative(path)}
        )
        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."""
    data = await self._request("GET", PATH_SCREENSHOTS)
    return data if isinstance(data, list) else []

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}/{_seg(core)}/{_seg(filename)}", parse_json=False
    )

async_delete_screenshot async

async_delete_screenshot(core: str, filename: str) -> None

Delete one saved screenshot.

Source code in src/mister_fpga/client.py

async def async_delete_screenshot(self, core: str, filename: str) -> None:
    """Delete one saved screenshot."""
    await self._request(
        "DELETE", f"{PATH_SCREENSHOTS}/{_seg(core)}/{_seg(filename)}"
    )

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}/{_seg(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 menu background mode.

Parameters:

Name	Type	Description	Default
mode	int	One of :data:~mister_fpga.const.BACKGROUND_MODES (0 none, 2 wallpaper, 4/6 horizontal bars, 8/10 vertical bars, 12 spectrum, 14 black).	required

Raises:

Type	Description
ValueError	For any other value (the server rejects odd modes).

Source code in src/mister_fpga/client.py

async def async_set_background_mode(self, mode: int) -> None:
    """Set the menu background mode.

    Args:
        mode: One of :data:`~mister_fpga.const.BACKGROUND_MODES` (0 none,
            2 wallpaper, 4/6 horizontal bars, 8/10 vertical bars,
            12 spectrum, 14 black).

    Raises:
        ValueError: For any other value (the server rejects odd modes).
    """
    if mode not in BACKGROUND_MODES:
        raise ValueError(f"invalid background mode {mode}")
    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}/{_seg(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}/{_seg(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}/{_seg(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 a token: a path, or a **command: such as **random:snes.

The text is base64url-encoded (no padding) as the /l/<data> endpoint requires; pass it in plain form.

Source code in src/mister_fpga/client.py

async def async_launch_token(self, data: str) -> None:
    """Launch a token: a path, or a ``**command:`` such as ``**random:snes``.

    The text is base64url-encoded (no padding) as the ``/l/<data>``
    endpoint requires; pass it in plain form.
    """
    encoded = base64.urlsafe_b64encode(data.encode()).decode().rstrip("=")
    await self._request("GET", f"{PATH_LAUNCH_TOKEN}/{encoded}")

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

Tap a Linux evdev KEY_* code (e.g. 28 = Enter).

A negative code sends Shift + abs(code).

Source code in src/mister_fpga/client.py

async def async_send_keyboard_raw(self, code: int) -> None:
    """Tap a Linux evdev ``KEY_*`` code (e.g. 28 = Enter).

    A negative code sends Shift + ``abs(code)``.
    """
    await self._request("POST", f"{PATH_KEYBOARD_RAW}/{code}")

async_get_screen async

async_get_screen() -> dict

Return the current framebuffer resolution {width, height}.

Source code in src/mister_fpga/client.py

async def async_get_screen(self) -> dict:
    """Return the current framebuffer resolution ``{width, height}``."""
    return await self._request("GET", PATH_SCREEN) or {}

async_get_playlog_status async

async_get_playlog_status() -> dict

PlayLog state: {installed, running, database, recents, startup}.

Source code in src/mister_fpga/client.py

async def async_get_playlog_status(self) -> dict:
    """PlayLog state: ``{installed, running, database, recents, startup}``."""
    return await self._request("GET", PATH_PLAYLOG_STATUS) or {}

async_get_playlog_summary async

async_get_playlog_summary(*, top: int | None = None, days: int | None = None) -> dict

Play-time summary: totals (seconds), streaks, top games/systems, per-day.

Source code in src/mister_fpga/client.py

async def async_get_playlog_summary(
    self, *, top: int | None = None, days: int | None = None
) -> dict:
    """Play-time summary: totals (seconds), streaks, top games/systems, per-day."""
    params = "&".join(
        f"{k}={v}" for k, v in (("top", top), ("days", days)) if v is not None
    )
    path = PATH_PLAYLOG_SUMMARY + (f"?{params}" if params else "")
    return await self._request("GET", path) or {}

async_get_playlog_games async

async_get_playlog_games(*, sort: str | None = None, limit: int | None = None, offset: int = 0) -> dict

Per-game play stats {"games": [...], "total"}.

Source code in src/mister_fpga/client.py

async def async_get_playlog_games(
    self, *, sort: str | None = None, limit: int | None = None, offset: int = 0
) -> dict:
    """Per-game play stats ``{"games": [...], "total"}``."""
    params = [f"offset={offset}"]
    if sort:
        params.append(f"sort={_seg(sort)}")
    if limit:
        params.append(f"limit={limit}")
    return (
        await self._request("GET", f"{PATH_PLAYLOG_GAMES}?{'&'.join(params)}")
        or {}
    )

async_get_playlog_sessions async

async_get_playlog_sessions(*, limit: int | None = None, offset: int = 0) -> dict

Recent play sessions {"sessions": [...], "total"}.

Source code in src/mister_fpga/client.py

async def async_get_playlog_sessions(
    self, *, limit: int | None = None, offset: int = 0
) -> dict:
    """Recent play sessions ``{"sessions": [...], "total"}``."""
    params = f"offset={offset}" + (f"&limit={limit}" if limit else "")
    return await self._request("GET", f"{PATH_PLAYLOG_SESSIONS}?{params}") or {}

async_get_remote_log async

async_get_remote_log() -> bytes

Download the Remote service log.

Source code in src/mister_fpga/client.py

async def async_get_remote_log(self) -> bytes:
    """Download the Remote service log."""
    return await self._request("GET", PATH_REMOTE_LOG, parse_json=False)

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	Running game as "<systemId>/<filename>" (or the bare core name for arcade), or None. This is not a filesystem path; do not pass it to :meth:MisterClient.async_launch_game.
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.
updated	str | None	Time of the last MiSTer update, not of this snapshot.
dns	str | None	<hostname>.local mDNS name (only when mDNS is enabled) — not a DNS server.
disk_total	int | None	Total SD-card space in bytes.
disk_used	int | None	Used SD-card space in bytes.
disk_free	int | None	Free SD-card space in bytes.
disks	list[dict]	Every disk reported by sysinfo (SD card, USB drives, ...).
core_session_time	int | None	Seconds the current core has been running (Remote >= 0.5, else None).
game_session_time	int | None	Seconds the current game has been running (Remote >= 0.5, else None).

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: Running game as ``"<systemId>/<filename>"`` (or the bare core
            name for arcade), or ``None``.  This is *not* a filesystem path;
            do not pass it to :meth:`MisterClient.async_launch_game`.
        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.
        updated: Time of the last MiSTer update, not of this snapshot.
        dns: ``<hostname>.local`` mDNS name (only when mDNS is enabled) — not a
            DNS server.
        disk_total: Total SD-card space in bytes.
        disk_used: Used SD-card space in bytes.
        disk_free: Free SD-card space in bytes.
        disks: Every disk reported by sysinfo (SD card, USB drives, ...).
        core_session_time: Seconds the current core has been running
            (Remote >= 0.5, else ``None``).
        game_session_time: Seconds the current game has been running
            (Remote >= 0.5, else ``None``).
    """

    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
    disks: list[dict] = field(default_factory=list)
    core_session_time: int | None = None
    game_session_time: 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
        self._ws: aiohttp.ClientWebSocketResponse | None = None

    @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:
                        self._ws = 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
                finally:
                    self._ws = None
                if self._stop:
                    break
                await asyncio.sleep(self._reconnect_delay)
        finally:
            if owns:
                await session.close()

    @property
    def connected(self) -> bool:
        """True while a WebSocket connection is open."""
        return self._ws is not None and not self._ws.closed

    async def async_send(self, message: str) -> None:
        """Send a raw command frame on the open connection.

        Server commands: ``getIndexStatus``, ``kbd:<name>``, ``kbdRaw:<code>``,
        ``kbdRawDown:<code>`` / ``kbdRawUp:<code>`` (hold a key), and on
        Remote >= 0.5 ``mouseMove``/``mousePos``/``mouseBtn``.  Unknown
        commands are answered with an ``invalid`` frame.

        Raises:
            ConnectionError: If not connected.
        """
        if not self.connected:
            raise ConnectionError("MiSTer WebSocket is not connected")
        await self._ws.send_str(message)

    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.

connected property

connected: bool

True while a WebSocket connection is open.

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:
                    self._ws = 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
            finally:
                self._ws = None
            if self._stop:
                break
            await asyncio.sleep(self._reconnect_delay)
    finally:
        if owns:
            await session.close()

async_send async

async_send(message: str) -> None

Send a raw command frame on the open connection.

Server commands: getIndexStatus, kbd:<name>, kbdRaw:<code>, kbdRawDown:<code> / kbdRawUp:<code> (hold a key), and on Remote >= 0.5 mouseMove/mousePos/mouseBtn. Unknown commands are answered with an invalid frame.

Raises:

Type	Description
ConnectionError	If not connected.

Source code in src/mister_fpga/websocket.py

async def async_send(self, message: str) -> None:
    """Send a raw command frame on the open connection.

    Server commands: ``getIndexStatus``, ``kbd:<name>``, ``kbdRaw:<code>``,
    ``kbdRawDown:<code>`` / ``kbdRawUp:<code>`` (hold a key), and on
    Remote >= 0.5 ``mouseMove``/``mousePos``/``mouseBtn``.  Unknown
    commands are answered with an ``invalid`` frame.

    Raises:
        ConnectionError: If not connected.
    """
    if not self.connected:
        raise ConnectionError("MiSTer WebSocket is not connected")
    await self._ws.send_str(message)

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; switching from another core clears the stale system and game fields (a launch from the menu sends gameRunning first, so that is kept).
• gameRunning — updates :attr:~mister_fpga.MisterStatus.game and :attr:~mister_fpga.MisterStatus.game_name (<system>/<file>, or the bare core name for arcade).
• menuNavigation — updates menu_path to FULLPATH/CURRENTPATH (see :func:split_menu_navigation for the parts).
• indexStatus — updates index_state (exists, in_progress); use :func:parse_index_status for the progress fields.

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`;
      switching from another core clears the stale system and game fields
      (a launch from the menu sends ``gameRunning`` first, so that is kept).
    * ``gameRunning`` — updates :attr:`~mister_fpga.MisterStatus.game` and
      :attr:`~mister_fpga.MisterStatus.game_name` (``<system>/<file>``, or the
      bare core name for arcade).
    * ``menuNavigation`` — updates *menu_path* to ``FULLPATH/CURRENTPATH``
      (see :func:`split_menu_navigation` for the parts).
    * ``indexStatus`` — updates *index_state* ``(exists, in_progress)``; use
      :func:`parse_index_status` for the progress fields.

    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,
            )
        if core == status.core:
            return status, menu_path, index_state
        if status.core is None:
            # On launch mrext sends gameRunning before coreRunning; keep the game.
            return replace(status, core=core), menu_path, index_state
        return (
            replace(
                status, core=core, system=None, system_name=None,
                game=None, game_name=None,
            ),
            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
        _, sep, name = rest.partition("/")
        if not sep:  # arcade: the bare core/MRA name
            name = rest
        game_name = name.rsplit(".", 1)[0] if sep else name
        return replace(status, game=rest, game_name=game_name), menu_path, index_state
    if prefix == "menuNavigation":
        full, current = split_menu_navigation(rest)
        path = "/".join(p for p in (full, current) if p) or None
        return status, path, index_state
    if prefix == "indexStatus":
        idx = parse_index_status(rest)
        return status, menu_path, (idx.exists, idx.indexing)
    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
        self._lock = asyncio.Lock()

    async def _ensure(self) -> None:
        if self._conn is not None:
            return
        self._conn = await asyncssh.connect(
            self.host,
            port=self.port,
            username=self.username,
            password=self.password,
            known_hosts=None,
        )

    async def async_run(self, command: str, timeout: int = 15) -> tuple[int, str]:
        """Run a command over the shared SSH connection.

        Returns ``(exit_status, stdout)``. Raises on connection/transport
        failure (callers that need best-effort behaviour must catch).
        """
        async with self._lock:
            try:
                await self._ensure()
                result = await self._conn.run(command, check=False, timeout=timeout)
            except Exception:
                self._conn = None
                raise
            return result.exit_status or 0, result.stdout or ""

    async def async_probe(self) -> dict:
        """Run the SSH probe command and return parsed telemetry."""
        try:
            _rc, out = await self.async_run(SSH_PROBE_CMD, timeout=10)
            return parse_ssh_probe(out)
        except Exception as err:  # noqa: BLE001 - SSH is best-effort; never break the HTTP integration
            _LOGGER.debug("MiSTer SSH probe failed: %s", err)
            return {}

    async def async_send_command(self, command: str) -> None:
        """Queue one command on ``/dev/MiSTer_cmd`` without ever blocking.

        Main reads the FIFO only while its loop is alive, so a plain ``echo``
        hangs forever when Main is stuck.  The write here is non-blocking and
        fails fast instead.

        Args:
            command: e.g. ``"volume 5"``, ``"volume mute"``,
                ``"load_core /media/fat/menu.rbf"``, ``"screenshot"``.

        Raises:
            ValueError: If *command* is empty or contains a newline.
            MisterMainUnresponsiveError: If the FIFO is full or missing.
        """
        if "\n" in command or not command.strip():
            raise ValueError(f"invalid MiSTer command: {command!r}")
        rc, out = await self.async_run(
            f"python3 -c {shlex.quote(_FIFO_WRITE_PY)} {shlex.quote(command)}",
            timeout=10,
        )
        if rc != 0:
            raise MisterMainUnresponsiveError(out.strip() or f"rc={rc}")

    async def async_set_volume(self, level: int) -> None:
        """Set master volume, 0-7 (7 loudest)."""
        if not 0 <= level <= 7:
            raise ValueError("volume must be 0-7")
        await self.async_send_command(f"volume {level}")

    async def async_set_mute(self, muted: bool) -> None:
        """Mute or unmute master audio."""
        await self.async_send_command("volume mute" if muted else "volume unmute")

    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_run async

async_run(command: str, timeout: int = 15) -> tuple[int, str]

Run a command over the shared SSH connection.

Returns (exit_status, stdout). Raises on connection/transport failure (callers that need best-effort behaviour must catch).

Source code in src/mister_fpga/ssh.py

async def async_run(self, command: str, timeout: int = 15) -> tuple[int, str]:
    """Run a command over the shared SSH connection.

    Returns ``(exit_status, stdout)``. Raises on connection/transport
    failure (callers that need best-effort behaviour must catch).
    """
    async with self._lock:
        try:
            await self._ensure()
            result = await self._conn.run(command, check=False, timeout=timeout)
        except Exception:
            self._conn = None
            raise
        return result.exit_status or 0, result.stdout or ""

async_probe async

async_probe() -> dict

Run the SSH probe command and return parsed telemetry.

Source code in src/mister_fpga/ssh.py

async def async_probe(self) -> dict:
    """Run the SSH probe command and return parsed telemetry."""
    try:
        _rc, out = await self.async_run(SSH_PROBE_CMD, timeout=10)
        return parse_ssh_probe(out)
    except Exception as err:  # noqa: BLE001 - SSH is best-effort; never break the HTTP integration
        _LOGGER.debug("MiSTer SSH probe failed: %s", err)
        return {}

async_send_command async

async_send_command(command: str) -> None

Queue one command on /dev/MiSTer_cmd without ever blocking.

Main reads the FIFO only while its loop is alive, so a plain echo hangs forever when Main is stuck. The write here is non-blocking and fails fast instead.

Parameters:

Name	Type	Description	Default
command	str	e.g. "volume 5", "volume mute", "load_core /media/fat/menu.rbf", "screenshot".	required

Raises:

Type	Description
ValueError	If command is empty or contains a newline.
MisterMainUnresponsiveError	If the FIFO is full or missing.

Source code in src/mister_fpga/ssh.py

async def async_send_command(self, command: str) -> None:
    """Queue one command on ``/dev/MiSTer_cmd`` without ever blocking.

    Main reads the FIFO only while its loop is alive, so a plain ``echo``
    hangs forever when Main is stuck.  The write here is non-blocking and
    fails fast instead.

    Args:
        command: e.g. ``"volume 5"``, ``"volume mute"``,
            ``"load_core /media/fat/menu.rbf"``, ``"screenshot"``.

    Raises:
        ValueError: If *command* is empty or contains a newline.
        MisterMainUnresponsiveError: If the FIFO is full or missing.
    """
    if "\n" in command or not command.strip():
        raise ValueError(f"invalid MiSTer command: {command!r}")
    rc, out = await self.async_run(
        f"python3 -c {shlex.quote(_FIFO_WRITE_PY)} {shlex.quote(command)}",
        timeout=10,
    )
    if rc != 0:
        raise MisterMainUnresponsiveError(out.strip() or f"rc={rc}")

async_set_volume async

async_set_volume(level: int) -> None

Set master volume, 0-7 (7 loudest).

Source code in src/mister_fpga/ssh.py

async def async_set_volume(self, level: int) -> None:
    """Set master volume, 0-7 (7 loudest)."""
    if not 0 <= level <= 7:
        raise ValueError("volume must be 0-7")
    await self.async_send_command(f"volume {level}")

async_set_mute async

async_set_mute(muted: bool) -> None

Mute or unmute master audio.

Source code in src/mister_fpga/ssh.py

async def async_set_mute(self, muted: bool) -> None:
    """Mute or unmute master audio."""
    await self.async_send_command("volume mute" if muted else "volume unmute")

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 eleven fields separated by |||. Any missing or malformed field is silently coerced to None, so output from an older probe still parses.

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.
dict	cmd_fifo_pending (int | None) — unread command-FIFO bytes.
dict	main_open_fds (int | None) — open descriptors of Main.
dict	volume (int | None) — 0-7, 7 loudest.
dict	muted (bool | None).
dict	game_id (str | None) — CRC32 or disc serial of the game.
dict	sam_active (bool | None) — Super Attract Mode is cycling.
dict	sam_game (str | None) — game SAM shows, e.g. Breakers (neogeo).
dict	cmd_fifo_pending above zero on two probes in a row means Main is
dict	hung (its loop also drains /dev/MiSTer_cmd). Some Main builds
dict	leak one descriptor per core load; the limit is 1024.

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 eleven fields separated
    by ``|||``.  Any missing or malformed field is silently coerced to
    ``None``, so output from an older probe still parses.

    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.
        * ``cmd_fifo_pending`` (``int | None``) — unread command-FIFO bytes.
        * ``main_open_fds`` (``int | None``) — open descriptors of Main.
        * ``volume`` (``int | None``) — 0-7, 7 loudest.
        * ``muted`` (``bool | None``).
        * ``game_id`` (``str | None``) — CRC32 or disc serial of the game.
        * ``sam_active`` (``bool | None``) — Super Attract Mode is cycling.
        * ``sam_game`` (``str | None``) — game SAM shows, e.g. ``Breakers (neogeo)``.

        ``cmd_fifo_pending`` above zero on two probes in a row means Main is
        hung (its loop also drains ``/dev/MiSTer_cmd``).  Some Main builds
        leak one descriptor per core load; the limit is 1024.
    """
    parts = [p.strip() for p in raw.split(_SEP)]
    while len(parts) < 11:
        parts.append("")
    core, uptime_s, load_s, mem_s, fw_s = parts[:5]
    fifo_s, fds_s, vol_s, gameid_s, sam_s, sam_game_s = parts[5:11]

    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
    volume, muted = parse_volume(_int(vol_s) if vol_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,
        "cmd_fifo_pending": _int(fifo_s) if fifo_s else None,
        "main_open_fds": _int(fds_s) if fds_s else None,
        "volume": volume,
        "muted": muted,
        "game_id": parse_game_id(gameid_s),
        "sam_active": bool(_int(sam_s)) if sam_s else None,
        "sam_game": sam_game_s or None,
    }

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

mister_fpga.MisterUnsupportedError

Bases: MisterConnectionError

The endpoint does not exist on this Remote version.

mrext Remote answers unknown /api routes (or a wrong HTTP method) with its web UI, 200 text/html, instead of a 404.

Source code in src/mister_fpga/client.py

class MisterUnsupportedError(MisterConnectionError):
    """The endpoint does not exist on this Remote version.

    mrext Remote answers unknown ``/api`` routes (or a wrong HTTP method) with
    its web UI, ``200 text/html``, instead of a 404.
    """

mister_fpga.MisterMainUnresponsiveError

Bases: Exception

Main_MiSTer is not draining /dev/MiSTer_cmd; the command was not queued.

Source code in src/mister_fpga/ssh.py

class MisterMainUnresponsiveError(Exception):
    """Main_MiSTer is not draining /dev/MiSTer_cmd; the command was not queued."""

mister_fpga.MisterGamepad

Client for the Companion Remote virtual controller.

Parameters:

Name	Type	Description	Default
host	str	MiSTer IP or hostname.	required
port	int	Daemon port (default 9191).	DEFAULT_GAMEPAD_PORT
path	str	WebSocket path (default /remote/v1).	DEFAULT_GAMEPAD_PATH
session	ClientSession | None	Optional shared aiohttp.ClientSession.	None
timeout	float	Seconds for the whole exchange (connect + replies).	5

Source code in src/mister_fpga/gamepad.py

class MisterGamepad:
    """Client for the Companion Remote virtual controller.

    Args:
        host: MiSTer IP or hostname.
        port: Daemon port (default 9191).
        path: WebSocket path (default ``/remote/v1``).
        session: Optional shared ``aiohttp.ClientSession``.
        timeout: Seconds for the whole exchange (connect + replies).
    """

    def __init__(
        self,
        host: str,
        port: int = DEFAULT_GAMEPAD_PORT,
        path: str = DEFAULT_GAMEPAD_PATH,
        *,
        session: aiohttp.ClientSession | None = None,
        timeout: float = 5,
    ) -> None:
        self.url = f"ws://{host}:{port}{path}"
        self.status_url = f"http://{host}:{port}/status"
        self._session = session
        self._timeout = timeout

    async def _with_session(self, func):
        if self._session is not None:
            return await func(self._session)
        async with aiohttp.ClientSession() as session:
            return await func(session)

    async def async_available(self) -> bool:
        """True when the daemon answers ``GET /status``."""

        async def _probe(session: aiohttp.ClientSession) -> bool:
            try:
                async with session.get(
                    self.status_url, timeout=aiohttp.ClientTimeout(total=self._timeout)
                ) as resp:
                    return resp.status == 200 and (await resp.json()).get("ok") is True
            except (aiohttp.ClientError, TimeoutError, ValueError):
                return False

        return await self._with_session(_probe)

    async def async_send(self, *commands: dict[str, Any]) -> list[dict]:
        """Send raw daemon commands over one connection and return the replies.

        Raises:
            MisterGamepadError: On connection failure or an ``ok: false`` reply.
        """

        async def _run(session: aiohttp.ClientSession) -> list[dict]:
            replies = []
            try:
                async with asyncio.timeout(self._timeout), session.ws_connect(
                    self.url
                ) as ws:
                    await ws.receive_json(timeout=self._timeout)  # hello
                    for command in commands:
                        await ws.send_str(json.dumps(command))
                        reply = await ws.receive_json(timeout=self._timeout)
                        if not reply.get("ok"):
                            raise MisterGamepadError(reply.get("message") or "rejected")
                        replies.append(reply)
            except (aiohttp.ClientError, TimeoutError, ValueError) as err:
                raise MisterGamepadError(f"{self.url}: {err}") from err
            return replies

        return await self._with_session(_run)

    async def async_press(self, button: str, action: str = "tap") -> None:
        """Press a controller button or D-pad direction.

        Args:
            button: One of :data:`GAMEPAD_BUTTONS`.
            action: ``"tap"`` (45 ms press), ``"down"`` or ``"up"``.  A held
                button is released when the connection closes, so ``down``
                only lasts for the call — use :meth:`async_send` for chords.
        """
        if button not in GAMEPAD_BUTTONS:
            raise ValueError(f"unknown button {button!r}")
        if action not in GAMEPAD_ACTIONS:
            raise ValueError(f"unknown action {action!r}")
        name = "mode" if button == "home" else button
        await self.async_send({"type": "controller", "name": name, "action": action})

    async def async_key(self, key: str, action: str = "tap") -> None:
        """Press a virtual keyboard key by evdev name, e.g. ``"KEY_F12"``."""
        if action not in GAMEPAD_ACTIONS:
            raise ValueError(f"unknown action {action!r}")
        await self.async_send(
            {"type": "keyboard", "key": key.upper(), "action": action}
        )

async_available async

async_available() -> bool

True when the daemon answers GET /status.

Source code in src/mister_fpga/gamepad.py

async def async_available(self) -> bool:
    """True when the daemon answers ``GET /status``."""

    async def _probe(session: aiohttp.ClientSession) -> bool:
        try:
            async with session.get(
                self.status_url, timeout=aiohttp.ClientTimeout(total=self._timeout)
            ) as resp:
                return resp.status == 200 and (await resp.json()).get("ok") is True
        except (aiohttp.ClientError, TimeoutError, ValueError):
            return False

    return await self._with_session(_probe)

async_send async

async_send(*commands: dict[str, Any]) -> list[dict]

Send raw daemon commands over one connection and return the replies.

Raises:

Type	Description
MisterGamepadError	On connection failure or an ok: false reply.

Source code in src/mister_fpga/gamepad.py

async def async_send(self, *commands: dict[str, Any]) -> list[dict]:
    """Send raw daemon commands over one connection and return the replies.

    Raises:
        MisterGamepadError: On connection failure or an ``ok: false`` reply.
    """

    async def _run(session: aiohttp.ClientSession) -> list[dict]:
        replies = []
        try:
            async with asyncio.timeout(self._timeout), session.ws_connect(
                self.url
            ) as ws:
                await ws.receive_json(timeout=self._timeout)  # hello
                for command in commands:
                    await ws.send_str(json.dumps(command))
                    reply = await ws.receive_json(timeout=self._timeout)
                    if not reply.get("ok"):
                        raise MisterGamepadError(reply.get("message") or "rejected")
                    replies.append(reply)
        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
            raise MisterGamepadError(f"{self.url}: {err}") from err
        return replies

    return await self._with_session(_run)

async_press async

async_press(button: str, action: str = 'tap') -> None

Press a controller button or D-pad direction.

Parameters:

Name	Type	Description	Default
button	str	One of :data:GAMEPAD_BUTTONS.	required
action	str	"tap" (45 ms press), "down" or "up". A held button is released when the connection closes, so down only lasts for the call — use :meth:async_send for chords.	'tap'

Source code in src/mister_fpga/gamepad.py

async def async_press(self, button: str, action: str = "tap") -> None:
    """Press a controller button or D-pad direction.

    Args:
        button: One of :data:`GAMEPAD_BUTTONS`.
        action: ``"tap"`` (45 ms press), ``"down"`` or ``"up"``.  A held
            button is released when the connection closes, so ``down``
            only lasts for the call — use :meth:`async_send` for chords.
    """
    if button not in GAMEPAD_BUTTONS:
        raise ValueError(f"unknown button {button!r}")
    if action not in GAMEPAD_ACTIONS:
        raise ValueError(f"unknown action {action!r}")
    name = "mode" if button == "home" else button
    await self.async_send({"type": "controller", "name": name, "action": action})

async_key async

async_key(key: str, action: str = 'tap') -> None

Press a virtual keyboard key by evdev name, e.g. "KEY_F12".

Source code in src/mister_fpga/gamepad.py

async def async_key(self, key: str, action: str = "tap") -> None:
    """Press a virtual keyboard key by evdev name, e.g. ``"KEY_F12"``."""
    if action not in GAMEPAD_ACTIONS:
        raise ValueError(f"unknown action {action!r}")
    await self.async_send(
        {"type": "keyboard", "key": key.upper(), "action": action}
    )

mister_fpga.MisterGamepadError

Bases: Exception

The daemon is unreachable or rejected a command.

Source code in src/mister_fpga/gamepad.py

class MisterGamepadError(Exception):
    """The daemon is unreachable or rejected a command."""

mister_fpga.MisterIndexStatus dataclass

Game-index state carried by indexStatus: frames.

Source code in src/mister_fpga/websocket.py

@dataclass(frozen=True)
class MisterIndexStatus:
    """Game-index state carried by ``indexStatus:`` frames."""

    exists: bool = False
    indexing: bool = False
    total_steps: int = 0
    current_step: int = 0
    description: str = ""

mister_fpga.parse_index_status

parse_index_status(rest: str) -> MisterIndexStatus

Parse the payload of indexStatus:<y|n>,<y|n>,<total>,<current>,<desc>.

Source code in src/mister_fpga/websocket.py

def parse_index_status(rest: str) -> MisterIndexStatus:
    """Parse the payload of ``indexStatus:<y|n>,<y|n>,<total>,<current>,<desc>``."""
    parts = rest.split(",", 4)
    parts += [""] * (5 - len(parts))

    def _int(value: str) -> int:
        try:
            return int(value)
        except ValueError:
            return 0

    return MisterIndexStatus(
        exists=parts[0] == "y",
        indexing=parts[1] == "y",
        total_steps=_int(parts[2]),
        current_step=_int(parts[3]),
        description=parts[4],
    )

mister_fpga.split_menu_navigation

split_menu_navigation(rest: str) -> tuple[str | None, str | None]

Split menuNavigation:<FULLPATH>:<CURRENTPATH> into its two parts.

FULLPATH is the folder open in the menu, CURRENTPATH the highlighted entry. Older senders without the separator yield (rest, None).

Source code in src/mister_fpga/websocket.py

def split_menu_navigation(rest: str) -> tuple[str | None, str | None]:
    """Split ``menuNavigation:<FULLPATH>:<CURRENTPATH>`` into its two parts.

    *FULLPATH* is the folder open in the menu, *CURRENTPATH* the highlighted
    entry.  Older senders without the separator yield ``(rest, None)``.
    """
    full, sep, current = rest.strip().partition(":")
    if not sep:
        return full or None, None
    return full or None, current or None

mister_fpga.parse_volume

parse_volume(raw: int | None) -> tuple[int | None, bool | None]

Decode /media/fat/config/Volume.dat into (level, muted).

Bits 0-2 hold the attenuation (0 = loudest, 7 = quietest) and bit 4 the mute flag. level is returned the other way round, 0-7 with 7 loudest, which is also what the volume <n> FIFO command expects.

Source code in src/mister_fpga/ssh.py

def parse_volume(raw: int | None) -> tuple[int | None, bool | None]:
    """Decode ``/media/fat/config/Volume.dat`` into ``(level, muted)``.

    Bits 0-2 hold the attenuation (0 = loudest, 7 = quietest) and bit 4 the
    mute flag. ``level`` is returned the other way round, 0-7 with 7 loudest,
    which is also what the ``volume <n>`` FIFO command expects.
    """
    if raw is None:
        return None, None
    return 7 - (raw & 0x07), bool(raw & 0x10)

mister_fpga.parse_ra_log

parse_ra_log(text: str) -> list[dict]

Turn /tmp/ra_debug.log lines into events.

odelot's Main only writes this log with debug=1 in retroachievements.cfg. game_id and game_title lines follow a === GAME IDENTIFIED === header and are merged into one game_identified event.

Returns:

Type	Description
list[dict]	Dicts with a type key (login, game_identified,
list[dict]	achievement, leaderboard, game_completed, game_unload)
list[dict]	plus the captured fields; numeric fields are ints.

Source code in src/mister_fpga/ra.py

def parse_ra_log(text: str) -> list[dict]:
    """Turn ``/tmp/ra_debug.log`` lines into events.

    odelot's Main only writes this log with ``debug=1`` in
    ``retroachievements.cfg``.  ``game_id`` and ``game_title`` lines follow a
    ``=== GAME IDENTIFIED ===`` header and are merged into one
    ``game_identified`` event.

    Returns:
        Dicts with a ``type`` key (``login``, ``game_identified``,
        ``achievement``, ``leaderboard``, ``game_completed``, ``game_unload``)
        plus the captured fields; numeric fields are ints.
    """
    events: list[dict] = []
    for raw in text.splitlines():
        line = _ANSI.sub("", raw).strip()
        for kind, pattern in _LOG_PATTERNS:
            match = pattern.search(line)
            if not match:
                continue
            data = {
                k: int(v) if v is not None and v.isdigit() else v
                for k, v in match.groupdict().items()
            }
            if kind == "game_id":
                events.append({"type": "game_identified", **data})
            elif kind == "game_title":
                if events and events[-1]["type"] == "game_identified":
                    events[-1].setdefault("title", data["title"])
            else:
                events.append({"type": kind, **data})
            break
    return events

mister_fpga.MisterRA

RetroAchievements local control sharing a MisterSSH connection.

Source code in src/mister_fpga/ra.py

class MisterRA:
    """RetroAchievements local control sharing a MisterSSH connection."""

    def __init__(self, ssh) -> None:
        self._ssh = ssh

    async def async_status(self) -> MisterRAStatus | None:
        """Probe RA state. Returns None only on SSH failure (best-effort)."""
        try:
            _rc, out = await self._ssh.async_run(RA_STATUS_CMD, timeout=15)
        except Exception as err:  # noqa: BLE001 - best-effort; never break the integration
            _LOGGER.debug("RA status probe failed: %s", err)
            return None
        return parse_ra_status(out)

    async def async_cores_on(self) -> None:
        await self._run_checked("bash /media/fat/Scripts/.ra/ra_on.sh")

    async def async_cores_off(self) -> None:
        await self._run_checked("bash /media/fat/Scripts/.ra/ra_off.sh")

    async def async_set_hardcore(self, enabled: bool) -> None:
        arg = "on" if enabled else "off"
        await self._run_checked(f"bash /media/fat/Scripts/.ra/ra_hardcore.sh {arg}")

    async def async_read_log(self, offset: int = 0) -> tuple[int, list[dict]]:
        """Read new RA log events written since *offset*.

        Returns:
            ``(new_offset, events)``.  Pass ``new_offset`` back next time.  When
            Main restarts the log is truncated; the offset then resets to 0.
        """
        cmd = (
            f"S=$(stat -c %s {RA_DEBUG_LOG} 2>/dev/null || echo 0); echo $S; "
            f"[ $S -ge {int(offset)} ] && O={int(offset)} || O=0; "
            f"tail -c +$((O+1)) {RA_DEBUG_LOG} 2>/dev/null | head -c 262144"
        )
        _rc, out = await self._ssh.async_run(cmd, timeout=15)
        size_s, _, body = out.partition("\n")
        try:
            size = int(size_s.strip())
        except ValueError:
            return offset, []
        start = offset if size >= offset else 0
        complete = body[: body.rfind("\n") + 1]  # keep partial line
        return start + len(complete.encode()), parse_ra_log(complete)

    async def async_set_debug(self, enabled: bool) -> None:
        """Turn RA debug logging on/off (applies on the next core load)."""
        value = "1" if enabled else "0"
        cfg = shlex.quote(RA_CFG)
        await self._run_checked(
            f"grep -q '^debug=' {cfg} && sed -i 's/^debug=.*/debug={value}/' {cfg}"
            f" || echo 'debug={value}' >> {cfg}"
        )

    async def _run_checked(self, command: str) -> None:
        rc, out = await self._ssh.async_run(command, timeout=30)
        if rc != 0:
            raise MisterRAError(f"command failed (rc={rc}): {command}\n{out}")

async_status async

async_status() -> MisterRAStatus | None

Probe RA state. Returns None only on SSH failure (best-effort).

Source code in src/mister_fpga/ra.py

async def async_status(self) -> MisterRAStatus | None:
    """Probe RA state. Returns None only on SSH failure (best-effort)."""
    try:
        _rc, out = await self._ssh.async_run(RA_STATUS_CMD, timeout=15)
    except Exception as err:  # noqa: BLE001 - best-effort; never break the integration
        _LOGGER.debug("RA status probe failed: %s", err)
        return None
    return parse_ra_status(out)

async_read_log async

async_read_log(offset: int = 0) -> tuple[int, list[dict]]

Read new RA log events written since offset.

Returns:

Type	Description
int	(new_offset, events). Pass new_offset back next time. When
list[dict]	Main restarts the log is truncated; the offset then resets to 0.

Source code in src/mister_fpga/ra.py

async def async_read_log(self, offset: int = 0) -> tuple[int, list[dict]]:
    """Read new RA log events written since *offset*.

    Returns:
        ``(new_offset, events)``.  Pass ``new_offset`` back next time.  When
        Main restarts the log is truncated; the offset then resets to 0.
    """
    cmd = (
        f"S=$(stat -c %s {RA_DEBUG_LOG} 2>/dev/null || echo 0); echo $S; "
        f"[ $S -ge {int(offset)} ] && O={int(offset)} || O=0; "
        f"tail -c +$((O+1)) {RA_DEBUG_LOG} 2>/dev/null | head -c 262144"
    )
    _rc, out = await self._ssh.async_run(cmd, timeout=15)
    size_s, _, body = out.partition("\n")
    try:
        size = int(size_s.strip())
    except ValueError:
        return offset, []
    start = offset if size >= offset else 0
    complete = body[: body.rfind("\n") + 1]  # keep partial line
    return start + len(complete.encode()), parse_ra_log(complete)

async_set_debug async

async_set_debug(enabled: bool) -> None

Turn RA debug logging on/off (applies on the next core load).

Source code in src/mister_fpga/ra.py

async def async_set_debug(self, enabled: bool) -> None:
    """Turn RA debug logging on/off (applies on the next core load)."""
    value = "1" if enabled else "0"
    cfg = shlex.quote(RA_CFG)
    await self._run_checked(
        f"grep -q '^debug=' {cfg} && sed -i 's/^debug=.*/debug={value}/' {cfg}"
        f" || echo 'debug={value}' >> {cfg}"
    )

mister_fpga.MisterRAWeb

Async client for the public RetroAchievements.org Web API.

Parameters:

Name	Type	Description	Default
username	str	RA account username (used for both z and u).	required
api_key	str	RA Web API key (y), from retroachievements.org/settings.	required
session	ClientSession | None	Optional shared aiohttp.ClientSession. When None the client creates and owns one; call :meth:async_close to release it.	None
timeout	int	Per-request timeout in seconds (default 15).	15

Source code in src/mister_fpga/ra_web.py

class MisterRAWeb:
    """Async client for the public RetroAchievements.org Web API.

    Args:
        username: RA account username (used for both ``z`` and ``u``).
        api_key: RA Web API key (``y``), from retroachievements.org/settings.
        session: Optional shared ``aiohttp.ClientSession``. When ``None`` the
            client creates and owns one; call :meth:`async_close` to release it.
        timeout: Per-request timeout in seconds (default ``15``).
    """

    def __init__(
        self,
        username: str,
        api_key: str,
        *,
        session: aiohttp.ClientSession | None = None,
        timeout: int = 15,
    ) -> None:
        self.username = username
        self._api_key = api_key
        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 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, endpoint: str, params: dict[str, Any]) -> Any:
        session = await self._get_session()
        url = f"{RA_WEB_API_BASE}{endpoint}"
        query = {"z": self.username, "y": self._api_key, "u": self.username, **params}
        try:
            async with session.get(
                url,
                params=query,
                timeout=aiohttp.ClientTimeout(total=self._timeout),
            ) as resp:
                resp.raise_for_status()
                text = await resp.text()
                try:
                    data = json.loads(text)
                except json.JSONDecodeError as err:
                    raise MisterRAWebError(
                        f"{endpoint} returned invalid JSON: {err}"
                    ) from err
                if isinstance(data, dict) and data.get("Error"):
                    raise MisterRAWebError(f"{endpoint} error: {data['Error']}")
                return data
        except (TimeoutError, aiohttp.ClientError) as err:
            _LOGGER.debug("RA Web request %s failed: %s", endpoint, err)
            raise MisterRAWebError(f"{endpoint} failed: {err}") from err

    async def async_get_rank_and_score(self) -> dict:
        return await self._request("API_GetUserRankAndScore.php", {})

    async def async_get_recently_played(
        self, count: int = RA_WEB_DEFAULT_RECENT_COUNT
    ) -> list:
        return await self._request(
            "API_GetUserRecentlyPlayedGames.php", {"c": count}
        )

    async def async_get_recent_achievements(
        self, minutes: int = RA_WEB_DEFAULT_ACHIEVEMENT_MINUTES
    ) -> list:
        return await self._request(
            "API_GetUserRecentAchievements.php", {"m": minutes}
        )

    async def async_fetch_stats(self) -> MisterRAWebStats:
        """Fetch all cloud stats concurrently and assemble them."""
        rank_raw, games_raw, ach_raw = await asyncio.gather(
            self.async_get_rank_and_score(),
            self.async_get_recently_played(),
            self.async_get_recent_achievements(),
        )
        hardcore, softcore, rank, total = parse_rank_and_score(rank_raw)
        games = parse_recently_played(games_raw)
        achievements = parse_recent_achievements(ach_raw)
        return MisterRAWebStats(
            hardcore_points=hardcore,
            softcore_points=softcore,
            rank=rank,
            total_ranked=total,
            current_game=games[0] if games else None,
            recent_games=games,
            last_achievement=achievements[0] if achievements else None,
        )

    async def async_validate(self) -> None:
        """Verify credentials with a lightweight call. Raises on failure."""
        await self.async_get_rank_and_score()

    async def async_get_badge_image(self, url: str) -> bytes:
        """Fetch a badge/icon image by absolute URL."""
        session = await self._get_session()
        try:
            async with session.get(
                url, timeout=aiohttp.ClientTimeout(total=self._timeout)
            ) as resp:
                resp.raise_for_status()
                return await resp.read()
        except (TimeoutError, aiohttp.ClientError) as err:
            raise MisterRAWebError(f"badge fetch failed: {err}") from err

async_close async

async_close() -> None

Close the HTTP session if this client created it.

Source code in src/mister_fpga/ra_web.py

async def async_close(self) -> None:
    """Close the 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_fetch_stats async

async_fetch_stats() -> MisterRAWebStats

Fetch all cloud stats concurrently and assemble them.

Source code in src/mister_fpga/ra_web.py

async def async_fetch_stats(self) -> MisterRAWebStats:
    """Fetch all cloud stats concurrently and assemble them."""
    rank_raw, games_raw, ach_raw = await asyncio.gather(
        self.async_get_rank_and_score(),
        self.async_get_recently_played(),
        self.async_get_recent_achievements(),
    )
    hardcore, softcore, rank, total = parse_rank_and_score(rank_raw)
    games = parse_recently_played(games_raw)
    achievements = parse_recent_achievements(ach_raw)
    return MisterRAWebStats(
        hardcore_points=hardcore,
        softcore_points=softcore,
        rank=rank,
        total_ranked=total,
        current_game=games[0] if games else None,
        recent_games=games,
        last_achievement=achievements[0] if achievements else None,
    )

async_validate async

async_validate() -> None

Verify credentials with a lightweight call. Raises on failure.

Source code in src/mister_fpga/ra_web.py

async def async_validate(self) -> None:
    """Verify credentials with a lightweight call. Raises on failure."""
    await self.async_get_rank_and_score()

async_get_badge_image async

async_get_badge_image(url: str) -> bytes

Fetch a badge/icon image by absolute URL.

Source code in src/mister_fpga/ra_web.py

async def async_get_badge_image(self, url: str) -> bytes:
    """Fetch a badge/icon image by absolute URL."""
    session = await self._get_session()
    try:
        async with session.get(
            url, timeout=aiohttp.ClientTimeout(total=self._timeout)
        ) as resp:
            resp.raise_for_status()
            return await resp.read()
    except (TimeoutError, aiohttp.ClientError) as err:
        raise MisterRAWebError(f"badge fetch failed: {err}") from err

mister_fpga.MisterRAWebStats dataclass

Aggregated RetroAchievements cloud player stats.

Source code in src/mister_fpga/const.py

@dataclass
class MisterRAWebStats:
    """Aggregated RetroAchievements cloud player stats."""

    hardcore_points: int = 0
    softcore_points: int = 0
    rank: int | None = None
    total_ranked: int | None = None
    current_game: RAGameProgress | None = None
    recent_games: list[RAGameProgress] = field(default_factory=list)
    last_achievement: RAAchievement | None = None

mister_fpga.RAGameProgress dataclass

Per-game achievement progress from the RA Web API.

Source code in src/mister_fpga/const.py

@dataclass
class RAGameProgress:
    """Per-game achievement progress from the RA Web API."""

    game_id: int
    title: str
    console: str
    num_achieved: int = 0
    num_possible: int = 0
    percent: float = 0.0
    last_played: str | None = None
    icon_url: str | None = None

mister_fpga.RAAchievement dataclass

A single unlocked achievement from the RA Web API.

Source code in src/mister_fpga/const.py

@dataclass
class RAAchievement:
    """A single unlocked achievement from the RA Web API."""

    title: str
    description: str
    points: int
    game_title: str
    date: str | None = None
    badge_url: str | None = None

mister_fpga.MisterRAWebError

Bases: Exception

A RetroAchievements Web API request failed (network, HTTP, or auth).

Source code in src/mister_fpga/ra_web.py

class MisterRAWebError(Exception):
    """A RetroAchievements Web API request failed (network, HTTP, or auth)."""