ExampleΒΆ

This is an example of using SNAPstack to send RPCs:

import argparse
import asyncio
import logging

from snapstack.interface.serial_wrapper import SerialType
from snapstack.snap import Snap

LOG = logging.getLogger(__name__)


async def main(args):

    LOG.info("Starting Snap instance")
    snap = Snap()
    snap.start()

    LOG.info(f"Opening serial port {args.port}")
    snap.open_serial(args.port, SerialType.PYSERIAL)

    LOG.info("Getting bridge node's address...")
    bridge = await snap.get_serial_bridge_address(args.port)
    print(f"Found bridge address: {bridge}")

    # If no target is specified, use the bridge node's address
    target = args.target or bridge
    LOG.info(f"RPC target: {target}")

    LOG.info("Calling getLq()...")
    responses = await snap.call_dmcast_rpc([target], b"getLq", ())
    result = responses[target]
    LOG.info(f"Result = {result}")
    if result:
        print(f"{target}.getLq() = {result.args_as(int)}")
    else:
        print(f"No response calling {target}.getLq()")

    LOG.info("Calling vmStat(6)...")
    responses = await snap.call_dmcast_rpc([target], b"vmStat", (6,), callback_name=b"tellVmStat")
    result = responses[target]
    LOG.info(f"Result = {result}")
    if result:
        print(f"{target}.vmStat(6) = {result.args_as(int, str)}")
    else:
        print(f"No response calling {target}.vmStat(6)")

    snap.close_serial(args.port)


if __name__ == "__main__":

    parser = argparse.ArgumentParser(description="Example of SNAPstack RPCs")

    parser.add_argument(
        "port",
        help="""Serial port for SNAP device, e.g.: COM3 (on Windows), /dev/snap1 (on an E12/E20 gateway), /dev/ttyUSB0 (on Linux), or /dev/tty.usbserial-A123456B (on macOS)""",
        default="/dev/snap1",
    )
    # The type here is to convert the target from a string to bytes.
    parser.add_argument(
        "target",
        help="SNAP address of the node you want to reach.  Defaults to the bridge node.",
        nargs="?",
        type=lambda x: x.encode(),
    )
    parser.add_argument(
        "-v", "--verbose", help="Increase verbosity of logs (-v for INFO, -vv for DEBUG)", action="count", default=0
    )

    args = parser.parse_args()

    # Convert count of verbose flags to log level
    log_level = [logging.WARN, logging.INFO, logging.DEBUG][min(args.verbose, 2)]

    logging.basicConfig(level=log_level)

    asyncio.run(main(args))