resonance.mcp.tools.dio_tools

resonance.mcp.tools.dio_tools

MCP tools for digital I/O (DIO) channels.

list_dio_channels() async

List all available DIO channel names.

Returns:

Type Description
dict

Response with channels list

Source code in src/resonance/mcp/tools/dio_tools.py
async def list_dio_channels() -> dict[str, Any]:
    """
    List all available DIO channel names.

    Returns
    -------
    dict
        Response with channels list
    """
    return DIOChannelsResponse(channels=list(dio)).model_dump()

get_dio_states(channels=None) async

Get current states for DIO channels.

Parameters:

Name Type Description Default
channels list[str] | None

List of channel names to get states for. If None or empty, gets all channels.

None

Returns:

Type Description
dict

Response with channel states (boolean values)

Raises:

Type Description
RuntimeError

If server communication fails

ValueError

If invalid channel names are provided

Source code in src/resonance/mcp/tools/dio_tools.py
async def get_dio_states(channels: list[str] | None = None) -> dict[str, Any]:
    """
    Get current states for DIO channels.

    Parameters
    ----------
    channels : list[str] | None, optional
        List of channel names to get states for. If None or empty, gets all channels.

    Returns
    -------
    dict
        Response with channel states (boolean values)

    Raises
    ------
    RuntimeError
        If server communication fails
    ValueError
        If invalid channel names are provided
    """
    try:
        await connection_manager.ensure_connected()
        beamline = await connection_manager.get_server()

        target = channels if channels else list(dio)
        result = await beamline.dio.read(*target)
        states = {chan: bool(val) for chan, val in result.items()}
        return DIOStatesResponse(states=states).model_dump()

    except (ConnectionError, RuntimeError, ValueError):
        raise
    except Exception as e:
        raise RuntimeError(f"Error getting DIO states: {e}") from e