Remote SSH Test for NMigen

nmigen-test.py

import os # TODO: Use posixpath directly for remote paths.
import socket

import nmigen
from nmigen_boards.test.blinky import *
from nmigen_boards.arty_a7 import *

import paramiko

# NMIGEN_REMOTE_root- username@IP_ADDRESS:/path/to/root (must be set- relative paths allowed)
# NMIGEN_REMOTE_port- Default: 22
# NMIGEN_REMOTE_ssh_dir- Default: look in ~/.ssh for id_rsa and known_hosts and config file.

# TODO- Leverage OpenSSH Config files.
def extract_credentials(ssh_addr):
    (username, rest) = ssh_addr.split("@")
    (server, root) = rest.split(":")
    return (username, server, root)

if __name__ == "__main__":
    # Input arguments to execute_remote()
    build_name = "build.zip" # Archive name
    local_dir = "build" # Extract archive name locally to dir?
    remote_dir = "build" # Remote dir where archive will be distracted.

    # From env vars.
    (user, server, server_root) = extract_credentials("william@192.168.1.168:Projects/FPGA/nmigen/remote") # os.environ["NMIGEN_REMOTE_root"]
    port = 22 # os.environ["NMIGEN_REMOTE_port"]
    ssh_dir = "C:\\msys64\\home\\William\\.ssh" # os.environ["NMIGEN_REMOTE_ssh_dir"]

    plan = ArtyA7Platform().build(Blinky(), do_build=False, do_program=False)
    plan.execute_local(local_dir, run_script=False)
    plan.archive(build_name)

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((server, 22))

    trans = paramiko.Transport(sock)
    trans.start_client()
    priv_key = paramiko.RSAKey.from_private_key_file(os.path.join(ssh_dir, "id_rsa"))
    trans.auth_publickey("william", priv_key)

    if trans.is_authenticated():
        with trans.open_sftp_client() as sftp:
            sftp.chdir(server_root)
            try:
                sftp.mkdir(remote_dir)
            except IOError as e:
                pass # mkdir fails if directory exists. This is fine.
            sftp.put(build_name, build_name) # Mirror local file layout

        with trans.open_session() as channel:
            channel.exec_command("cd {} && unzip -qo ../build.zip && bash -l ./build_top.sh".format(os.path.join(server_root, remote_dir)))

            # Show the output from the server while products are built.
            while not channel.exit_status_ready():
                print(channel.recv(1024).decode("utf-8"), end="")

            # Drain the remaining contents of the channel.
            buf = channel.recv(1024)
            while buf:
                print(buf.decode("utf-8"), end="")
                buf = channel.recv(1024)

            if channel.recv_stderr_ready():
                buf = channel.recv_stderr(1024)
                while buf:
                    print(buf.decode("utf-8"), end="")
                    buf = channel.recv_stderr(1024)

        # Remote build products
        with trans.open_sftp_client() as sftp:
            sftp.chdir(os.path.join(server_root))
            # Use "file" and read into string in RemoteBuildProducts.
            # "top.bin, etc" is input argument to RemoteBuildProducts.get()
            sftp.get(os.path.join(remote_dir, "top.bin"), os.path.join(local_dir, "top.bin"))
            sftp.get(os.path.join(remote_dir, "top.bit"), os.path.join(local_dir, "top.bit"))