mcdosbox-x/src/mcdosbox_x/tools/control.py

"""Control tools: pause, resume, reset, savestate, loadstate."""

from typing import Any

from .peripheral import _get_qmp_port, _qmp_command


def pause() -> dict:
    """Pause DOSBox-X emulation.

    Stops the emulator at the current instruction. Use resume() to continue.
    This is different from GDB breakpoints - it's an immediate halt.

    Returns:
        Pause status
    """
    result = _qmp_command("localhost", _get_qmp_port(), "stop")

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    return {
        "success": True,
        "message": "Emulator paused",
    }


def resume() -> dict:
    """Resume DOSBox-X emulation after pause.

    Continues emulator execution after a pause() call.

    Returns:
        Resume status
    """
    result = _qmp_command("localhost", _get_qmp_port(), "cont")

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    return {
        "success": True,
        "message": "Emulator resumed",
    }


def reset(dos_only: bool = False) -> dict:
    """Reset DOSBox-X system.

    Args:
        dos_only: If True, only reset DOS (warm boot). If False, full system reset.

    Returns:
        Reset status
    """
    args = {}
    if dos_only:
        args["dos"] = True

    result = _qmp_command("localhost", _get_qmp_port(), "system_reset", args if args else None)

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    return {
        "success": True,
        "message": "DOS reset" if dos_only else "System reset",
        "reset_type": "dos" if dos_only else "full",
    }


def savestate(filename: str) -> dict:
    """Save DOSBox-X emulator state to file.

    Creates a snapshot of the entire emulator state that can be loaded later.
    Useful for creating checkpoints during reverse engineering.

    Args:
        filename: Path to save state file (relative to DOSBox working dir)

    Returns:
        Save status with file path

    Example:
        savestate("checkpoint1.sav")
    """
    result = _qmp_command(
        "localhost", _get_qmp_port(), "savestate", {"file": filename}, timeout=30.0
    )

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    response = {
        "success": True,
        "message": f"State saved to {filename}",
        "filename": filename,
    }

    # Include any return data from QMP
    if "return" in result:
        ret = result["return"]
        if isinstance(ret, dict):
            response.update(ret)

    return response


def loadstate(filename: str) -> dict:
    """Load DOSBox-X emulator state from file.

    Restores a previously saved snapshot. The emulator will be in exactly
    the same state as when the savestate was created.

    Args:
        filename: Path to state file to load

    Returns:
        Load status

    Example:
        loadstate("checkpoint1.sav")
    """
    result = _qmp_command(
        "localhost", _get_qmp_port(), "loadstate", {"file": filename}, timeout=30.0
    )

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    response = {
        "success": True,
        "message": f"State loaded from {filename}",
        "filename": filename,
    }

    if "return" in result:
        ret = result["return"]
        if isinstance(ret, dict):
            response.update(ret)

    return response


def memdump(address: str, length: int, filename: str | None = None) -> dict:
    """Dump memory region to file or return as data.

    Uses QMP memdump for efficient binary memory extraction.
    More efficient than GDB memory_read for large regions.

    Args:
        address: Start address in hex format (e.g., "0x0000", "1234:5678")
        length: Number of bytes to dump
        filename: Optional file to save dump (returns base64 if not specified)

    Returns:
        Memory dump result with either file path or base64 data

    Example:
        memdump("0xB8000", 4000)  # Dump VGA text buffer
        memdump("CS:IP", 256, "code_dump.bin")
    """

    # Parse segment:offset if provided
    if ":" in address:
        seg, off = address.split(":")
        flat_addr = (int(seg, 16) << 4) + int(off, 16)
    elif address.startswith("0x"):
        flat_addr = int(address, 16)
    else:
        flat_addr = int(address, 16)

    args: dict[str, Any] = {
        "address": flat_addr,
        "length": length,
    }
    if filename:
        args["filename"] = filename

    result = _qmp_command("localhost", _get_qmp_port(), "memdump", args, timeout=30.0)

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    response = {
        "success": True,
        "address": f"0x{flat_addr:X}",
        "length": length,
    }

    if "return" in result:
        ret = result["return"]
        if isinstance(ret, dict):
            if "file" in ret:
                response["filename"] = ret["file"]
                response["message"] = f"Memory dumped to {ret['file']}"
            if "data" in ret:
                # Return base64 data
                response["data_base64"] = ret["data"]
                response["message"] = f"Dumped {length} bytes"

    return response


def query_status() -> dict:
    """Query detailed DOSBox-X emulator status via QMP.

    Returns more detailed status than the GDB-based status() tool,
    including pause state and debugger activity.

    Returns:
        Detailed emulator status
    """
    result = _qmp_command("localhost", _get_qmp_port(), "query-status")

    if "error" in result:
        return {
            "success": False,
            "error": result.get("error", "Unknown error"),
        }

    response = {
        "success": True,
    }

    if "return" in result:
        ret = result["return"]
        if isinstance(ret, dict):
            response["running"] = ret.get("running", False)
            response["paused"] = not ret.get("running", False)
            response["debugger_active"] = ret.get("debugger", False)
            if "reason" in ret:
                response["pause_reason"] = ret["reason"]

    return response