From df674238979a0e028de370a123a04038279dea25 Mon Sep 17 00:00:00 2001 From: Dustin Spicuzza Date: Fri, 5 Jan 2024 05:27:00 -0500 Subject: [PATCH] Refactor installer - Make RobotpyInstaller object able to do all the things for future extensions - Store team number in .wpilib/wpilib_preferences.json for vscode compat - Remove CLI command, use robotpy entry point instead --- robotpy_installer/cli_deploy.py | 38 +- robotpy_installer/cli_deploy_info.py | 53 +- robotpy_installer/cli_installer.py | 350 ++++++++++ robotpy_installer/cli_undeploy.py | 9 +- robotpy_installer/installer.py | 873 +++++++++--------------- robotpy_installer/sshcontroller.py | 133 ++-- robotpy_installer/utils.py | 15 +- robotpy_installer/wpilib_preferences.py | 85 +++ setup.cfg | 2 +- tests/test_basic.py | 3 +- 10 files changed, 911 insertions(+), 650 deletions(-) create mode 100644 robotpy_installer/cli_installer.py create mode 100644 robotpy_installer/wpilib_preferences.py diff --git a/robotpy_installer/cli_deploy.py b/robotpy_installer/cli_deploy.py index 3697170..9383cd2 100644 --- a/robotpy_installer/cli_deploy.py +++ b/robotpy_installer/cli_deploy.py @@ -17,7 +17,7 @@ from os.path import join, splitext from . import sshcontroller -from .utils import print_err, yesno +from .utils import handle_cli_error, print_err, yesno import logging @@ -107,6 +107,7 @@ def __init__(self, parser: argparse.ArgumentParser): help="If specified, don't do a DNS lookup, allow ssh et al to do it instead", ) + @handle_cli_error def run( self, main_file: pathlib.Path, @@ -158,32 +159,23 @@ def run( # upload all files in the robot.py source directory robot_filename = main_file.name - cfg_filename = project_path / ".deploy_cfg" if not large and not self._check_large_files(project_path): return 1 - hostname_or_team = robot or team - - try: - with sshcontroller.ssh_from_cfg( - cfg_filename, - username="lvuser", - password="", - hostname=hostname_or_team, - no_resolve=no_resolve, - ) as ssh: - if not self._check_requirements(ssh, no_version_check): - return 1 - - if not self._do_deploy( - ssh, debug, nc, nc_ds, robot_filename, project_path - ): - return 1 - - except sshcontroller.SshExecError as e: - print_err("ERROR:", str(e)) - return 1 + with sshcontroller.ssh_from_cfg( + project_path, + main_file, + username="lvuser", + password="", + robot_or_team=robot or team, + no_resolve=no_resolve, + ) as ssh: + if not self._check_requirements(ssh, no_version_check): + return 1 + + if not self._do_deploy(ssh, debug, nc, nc_ds, robot_filename, project_path): + return 1 print("\nSUCCESS: Deploy was successful!") return 0 diff --git a/robotpy_installer/cli_deploy_info.py b/robotpy_installer/cli_deploy_info.py index 1490e65..a83f328 100644 --- a/robotpy_installer/cli_deploy_info.py +++ b/robotpy_installer/cli_deploy_info.py @@ -7,6 +7,7 @@ from . import sshcontroller +from .utils import handle_cli_error from .utils import print_err @@ -33,8 +34,10 @@ def __init__(self, parser: argparse.ArgumentParser): help="If specified, don't do a DNS lookup, allow ssh et al to do it instead", ) + @handle_cli_error def run( self, + project_path: pathlib.Path, main_file: pathlib.Path, robot: typing.Optional[str], team: typing.Optional[int], @@ -47,34 +50,26 @@ def run( ) return 1 - cfg_filename = main_file.parent / ".deploy_cfg" - - hostname_or_team = robot or team - - try: - with sshcontroller.ssh_from_cfg( - cfg_filename, - username="lvuser", - password="", - hostname=hostname_or_team, - no_resolve=no_resolve, - ) as ssh: - result = ssh.exec_cmd( - ( - "[ -f /home/lvuser/py/deploy.json ] && " - "cat /home/lvuser/py/deploy.json || " - "echo {}" - ), - get_output=True, - ) - if not result.stdout: - print("{}") - else: - data = json.loads(result.stdout) - print(json.dumps(data, indent=2, sort_keys=True)) - - except sshcontroller.SshExecError as e: - print_err("ERROR:", str(e)) - return 1 + with sshcontroller.ssh_from_cfg( + project_path, + main_file, + username="lvuser", + password="", + robot_or_team=robot or team, + no_resolve=no_resolve, + ) as ssh: + result = ssh.exec_cmd( + ( + "[ -f /home/lvuser/py/deploy.json ] && " + "cat /home/lvuser/py/deploy.json || " + "echo {}" + ), + get_output=True, + ) + if not result.stdout: + print("{}") + else: + data = json.loads(result.stdout) + print(json.dumps(data, indent=2, sort_keys=True)) return 0 diff --git a/robotpy_installer/cli_installer.py b/robotpy_installer/cli_installer.py new file mode 100644 index 0000000..a2573b7 --- /dev/null +++ b/robotpy_installer/cli_installer.py @@ -0,0 +1,350 @@ +import argparse +import pathlib +import shutil +import typing + +from .utils import handle_cli_error + +from .installer import ( + InstallerException, + RobotpyInstaller, + _IS_BETA, +) +from .utils import yesno + + +def _add_ssh_options(parser: argparse.ArgumentParser): + parser.add_argument( + "--robot", + help="Specify the robot hostname or team number", + ) + + parser.add_argument( + "--ignore-image-version", + action="store_true", + default=False, + help="Ignore RoboRIO image version", + ) + + +# +# installer cache +# + + +class InstallerCacheLocation: + """Print cache location""" + + def __init__(self, parser: argparse.ArgumentParser) -> None: + pass + + @handle_cli_error + def run(self): + installer = RobotpyInstaller(log_startup=False) + print(installer.cache_root) + + +class InstallerCacheRm: + """Delete all cached files""" + + def __init__(self, parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "-f", + "--force", + action="store_true", + default=False, + help="Force removal without asking", + ) + + @handle_cli_error + def run(self, force: bool): + installer = RobotpyInstaller(log_startup=False) + if force or yesno(f"Really delete {installer.cache_root}?"): + shutil.rmtree(installer.cache_root) + + +class InstallerCache: + """ + Installer cache management + """ + + subcommands = [ + ("location", InstallerCacheLocation), + ("rm", InstallerCacheRm), + ] + + +# +# Installer python +# + + +class InstallerDownloadPython: + """ + Downloads Python for RoboRIO + + You must be connected to the internet for this to work. + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--use-certifi", + action="store_true", + default=False, + help="Use SSL certificates from certifi", + ) + + def run(self, use_certifi: bool): + installer = RobotpyInstaller() + installer.download_python(use_certifi) + + +class InstallerInstallPython: + """ + Installs Python on a RoboRIO + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + _add_ssh_options(parser) + + @handle_cli_error + def run( + self, + project_path: pathlib.Path, + main_file: pathlib.Path, + ignore_image_version: bool, + robot: typing.Optional[str], + ): + installer = RobotpyInstaller() + with installer.connect_to_robot( + project_path=project_path, + main_file=main_file, + robot_or_team=robot, + ignore_image_version=ignore_image_version, + ): + installer.install_python() + + +class InstallerUninstallPython: + """ + Uninstall Python from a RoboRIO + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + _add_ssh_options(parser) + + @handle_cli_error + def run( + self, + project_path: pathlib.Path, + main_file: pathlib.Path, + ignore_image_version: bool, + robot: typing.Optional[str], + ): + installer = RobotpyInstaller() + with installer.connect_to_robot( + project_path=project_path, + main_file=main_file, + robot_or_team=robot, + ignore_image_version=ignore_image_version, + ): + installer.uninstall_python() + + +# +# Installer pip things +# + + +def common_pip_options( + parser: argparse.ArgumentParser, +): + parser.add_argument( + "--no-deps", + action="store_true", + default=False, + help="Don't install package dependencies", + ) + + parser.add_argument( + "--pre", + action="store_true", + default=_IS_BETA, + help="Include pre-release and development versions", + ) + + parser.add_argument( + "--requirements", + "-r", + action="append", + type=pathlib.Path, + default=[], + help="Install from the given requirements file. This option can be used multiple times.", + ) + + parser.add_argument( + "packages", + nargs="*", + help="Packages to be processed", + ) + + +class InstallerDownload: + """ + Specify Python package(s) to download, and store them in the cache. + + You must be connected to the internet for this to work. + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + common_pip_options(parser) + + @handle_cli_error + def run( + self, + no_deps: bool, + pre: bool, + requirements: typing.Tuple[str], + packages: typing.Tuple[str], + ): + installer = RobotpyInstaller() + installer.pip_download(no_deps, pre, requirements, packages) + + +class InstallerInstall: + """ + Installs Python package(s) on a RoboRIO. + + The package must already been downloaded with the 'download' command first. + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--force-reinstall", + action="store_true", + default=False, + help="When upgrading, reinstall all packages even if they are already up-to-date.", + ) + + parser.add_argument( + "--ignore-installed", + "-I", + action="store_true", + default=False, + help="Ignore the installed packages (reinstalling instead)", + ) + + common_pip_options(parser) + _add_ssh_options(parser) + + @handle_cli_error + def run( + self, + project_path: pathlib.Path, + main_file: pathlib.Path, + ignore_image_version: bool, + robot: typing.Optional[str], + force_reinstall: bool, + ignore_installed: bool, + no_deps: bool, + pre: bool, + requirements: typing.Tuple[str], + packages: typing.Tuple[str], + ): + if len(requirements) == 0 and len(packages) == 0: + raise InstallerException( + "You must give at least one requirement to install" + ) + + installer = RobotpyInstaller() + with installer.connect_to_robot( + project_path=project_path, + main_file=main_file, + robot_or_team=robot, + ignore_image_version=ignore_image_version, + ): + installer.pip_install( + force_reinstall, ignore_installed, no_deps, pre, requirements, packages + ) + + +class InstallerList: + """ + Lists Python packages present on RoboRIO + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + _add_ssh_options(parser) + + @handle_cli_error + def run( + self, + project_path: pathlib.Path, + main_file: pathlib.Path, + ignore_image_version: bool, + robot: typing.Optional[str], + ): + installer = RobotpyInstaller() + with installer.connect_to_robot( + project_path=project_path, + main_file=main_file, + robot_or_team=robot, + ignore_image_version=ignore_image_version, + log_disk_usage=False, + ): + installer.pip_list() + + +class InstallerUninstall: + """ + Uninstall Python packages from a RoboRIO + """ + + def __init__(self, parser: argparse.ArgumentParser) -> None: + _add_ssh_options(parser) + + parser.add_argument( + "packages", + nargs="*", + help="Packages to be processed", + ) + + @handle_cli_error + def run( + self, + project_path: pathlib.Path, + main_file: pathlib.Path, + ignore_image_version: bool, + robot: typing.Optional[str], + packages: typing.List[str], + ): + installer = RobotpyInstaller() + with installer.connect_to_robot( + project_path=project_path, + main_file=main_file, + robot_or_team=robot, + ignore_image_version=ignore_image_version, + ): + installer.pip_uninstall(packages) + + +# +# Installer command +# + + +class Installer: + """ + Manage RobotPy on your RoboRIO + """ + + subcommands = [ + ("cache", InstallerCache), + ("download", InstallerDownload), + ("download-python", InstallerDownloadPython), + ("install", InstallerInstall), + ("install-python", InstallerInstallPython), + ("list", InstallerList), + ("uninstall", InstallerUninstall), + ("uninstall-python", InstallerUninstallPython), + ] diff --git a/robotpy_installer/cli_undeploy.py b/robotpy_installer/cli_undeploy.py index 165534d..b0d1b3b 100644 --- a/robotpy_installer/cli_undeploy.py +++ b/robotpy_installer/cli_undeploy.py @@ -42,6 +42,7 @@ def __init__(self, parser: argparse.ArgumentParser): def run( self, + project_path: pathlib.Path, main_file: pathlib.Path, robot: typing.Optional[str], team: typing.Optional[int], @@ -54,7 +55,6 @@ def run( file=sys.stderr, ) return 1 - cfg_filename = main_file.parent / ".deploy_cfg" if not yes: if not yesno( @@ -62,14 +62,13 @@ def run( ): return 1 - hostname_or_team = robot or team - try: with sshcontroller.ssh_from_cfg( - cfg_filename, + project_path, + main_file, username="lvuser", password="", - hostname=hostname_or_team, + robot_or_team=robot or team, no_resolve=no_resolve, ) as ssh: # first, turn off the running program diff --git a/robotpy_installer/installer.py b/robotpy_installer/installer.py index 2ae2ad4..a7f788d 100755 --- a/robotpy_installer/installer.py +++ b/robotpy_installer/installer.py @@ -4,15 +4,11 @@ import logging import pathlib import re -import shutil import subprocess import sys from urllib.parse import urlparse import typing -import click -from click import argument, option, group, pass_context, pass_obj, ClickException - from os.path import basename from .version import version as __version__ @@ -45,35 +41,8 @@ logger = logging.getLogger("robotpy.installer") -class RobotpyInstaller: - def __init__(self, cache_root: pathlib.Path, cfgroot: pathlib.Path): - self.cache_root = cache_root - self.pip_cache = cache_root / "pip_cache" - self.opkg_cache = cache_root / "opkg_cache" - - self.cfg_filename = cfgroot / ".installer_config" - - def log_startup(self) -> None: - logger.info("RobotPy Installer %s", __version__) - logger.info("-> caching files at %s", self.cache_root) - - def get_ssh(self, robot: typing.Optional[str]) -> SshController: - try: - return ssh_from_cfg( - self.cfg_filename, username="admin", password="", hostname=robot - ) - except Error as e: - raise ClickException(str(e)) from e - - def start_cache(self, ssh: SshController) -> CacheServer: - cache = CacheServer(ssh, self.cache_root) - cache.start() - return cache - - -# -# Helpers -# +class InstallerException(Error): + pass @contextlib.contextmanager @@ -81,225 +50,99 @@ def catch_ssh_error(msg: str): try: yield except SshExecError as e: - raise ClickException(f"{msg}: {e}") - + raise InstallerException(f"{msg}: {e}") -def remove_legacy_components(ssh: SshController): - # (remove in 2022) check for old robotpy components - # -> only removes opkg components, pip will take care of the rest - with catch_ssh_error("check for old RobotPy"): - result = ssh.check_output("opkg list-installed python38*").strip() +class RobotpyInstaller: + def __init__(self, *, log_startup: bool = True): + self.cache_root = pathlib.Path.home() / "wpilib" / _WPILIB_YEAR / "robotpy" + self.pip_cache = self.cache_root / "pip_cache" + self.opkg_cache = self.cache_root / "opkg_cache" + + self._ssh: typing.Optional[SshController] = None + self._cache_server: typing.Optional[CacheServer] = None + + self._image_version_ok = False + self._robot_pip_ok = False + + if log_startup: + logger.info("RobotPy Installer %s", __version__) + logger.info("-> caching files at %s", self.cache_root) + + @contextlib.contextmanager + def connect_to_robot( + self, + *, + project_path: pathlib.Path, + main_file: pathlib.Path, + robot_or_team: typing.Union[None, str, int] = None, + ignore_image_version: bool = False, + log_disk_usage: bool = True, + no_resolve: bool = False, + ): + ssh = ssh_from_cfg( + project_path, + main_file, + username="admin", + password="", + robot_or_team=robot_or_team, + no_resolve=no_resolve, + ) - if result != "": - packages = [line.split()[0] for line in result.splitlines()] + with ssh: + self._ssh = ssh - print("RobotPy 2020 components detected!") - for package in packages: - print("-", package) + self.ensure_image_version(ignore_image_version) - if not click.confirm("Uninstall?"): - raise ClickException("installer cannot continue") + if log_disk_usage: + self.show_disk_space() - with catch_ssh_error("uninstall old RobotPy"): - ssh.exec_cmd(f"opkg remove {' '.join(packages)}", print_output=True) + yield + if log_disk_usage: + self.show_disk_space() -def show_disk_space( - ssh: SshController, -) -> typing.Tuple[str, str, str]: - with catch_ssh_error("checking free space"): - result = ssh.check_output("df -h / | tail -n 1") + self._ssh = None - _, size, used, _, pct, _ = result.strip().split() - logger.info("-> RoboRIO disk usage %s/%s (%s full)", used, size, pct) + @property + def cache_server(self) -> CacheServer: + """Only access inside connect_to_robot context""" + if not self._cache_server: + self._cache_server = CacheServer(self.ssh, self.cache_root) + self._cache_server.start() - return size, used, pct + return self._cache_server + @property + def ssh(self) -> SshController: + """Only access inside connect_to_robot context""" + if self._ssh is None: + raise RuntimeError("internal error") + return self._ssh -def roborio_checks( - ssh: SshController, - ignore_image_version: bool, - pip_check: bool = False, -): # - # Image version check - # - - with catch_ssh_error("retrieving image version"): - result = ssh.check_output( - "grep IMAGEVERSION /etc/natinst/share/scs_imagemetadata.ini", - ) - - roborio_match = re.match(r'IMAGEVERSION = "(FRC_)?roboRIO_(.*)"', result.strip()) - roborio2_match = re.match(r'IMAGEVERSION = "(FRC_)?roboRIO2_(.*)"', result.strip()) - - if roborio_match: - version = roborio_match.group(2) - images = _ROBORIO_IMAGES - name = "RoboRIO" - elif roborio2_match: - version = roborio2_match.group(2) - images = _ROBORIO2_IMAGES - name = "RoboRIO 2" - else: - version = "" - images = [ - f"({_ROBORIO_IMAGES[-1]} | {_ROBORIO2_IMAGES[-1]})", - ] - name = "RoboRIO (1 | 2)" - - logger.info(f"-> {name} image version: {version}") - - if not ignore_image_version and version not in images: - raise ClickException( - f"{name} image {images[-1]} is required!\n" - "\n" - "See https://docs.wpilib.org/en/stable/docs/zero-to-robot/step-3/imaging-your-roborio.html\n" - "for information about upgrading the RoboRIO image.\n" - "\n" - "Use --ignore-image-version to install anyways" - ) - + # Utilities # - # Free space check.. maybe in the future we'll use this to not accidentally - # fill the user's disk, but it'd be annoying to figure out - # - - show_disk_space(ssh) - - # - # Ensure that pip is installed - # - - if pip_check: - with catch_ssh_error("checking for pip3"): - if ssh.exec_cmd("[ -x /usr/local/bin/pip3 ]").returncode != 0: - raise ClickException( - inspect.cleandoc( - """ - pip3 not found on RoboRIO, did you install python? - - Use the 'download-python' and 'install-python' commands first! - """ - ) - ) - - # Use pip stub to override the wheel platform on roborio - with catch_ssh_error("copying pip stub"): - from . import _pipstub - - stub_fp = io.BytesIO() - stub_fp.write(b"#!/usr/local/bin/python3\n\n") - stub_fp.write(inspect.getsource(_pipstub).encode("utf-8")) - stub_fp.seek(0) - - ssh.sftp_fp(stub_fp, _PIP_STUB_PATH) - ssh.exec_cmd(f"chmod +x {_PIP_STUB_PATH}", check=True) - - -# -# Click-based CLI -# - - -def _make_ssl_context(use_certifi: bool): - if not use_certifi: - return None - - try: - import certifi # type: ignore - except ImportError: - raise click.ClickException( - "certifi is not installed, please install it via `pip install certifi`" - ) - - import ssl - - return ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certifi.where()) - - -@group() -@pass_context -def installer(ctx: click.Context): - """ - RobotPy installer utility - """ - - log_datefmt = "%H:%M:%S" - log_format = "%(asctime)s:%(msecs)03d %(levelname)-8s: %(name)-20s: %(message)s" - - logging.basicConfig(datefmt=log_datefmt, format=log_format, level=logging.INFO) - - cache_root = pathlib.Path.home() / "wpilib" / _WPILIB_YEAR / "robotpy" - cfg_root = pathlib.Path.cwd() - - # This becomes the first argument to any cli command with the @pass_obj decorator - ctx.obj = RobotpyInstaller(cache_root, cfg_root) - - -def _common_ssh_options(f): - f = option("--robot", help="Specify the robot hostname or team number")(f) - f = option( - "--ignore-image-version", is_flag=True, help="Ignore RoboRIO image version" - )(f) - return f - - -# -# Cache management -# - - -@installer.group() -def cache(): - """Cache management""" - - -@cache.command() -@pass_obj -def location(installer: RobotpyInstaller): - """Print cache location""" - print(installer.cache_root) - - -@cache.command() -@option("-f", "--force", is_flag=True, help="Force removal without asking") -@pass_obj -def rm(installer: RobotpyInstaller, force: bool): - """Delete all cached files""" - if force or click.confirm(f"Really delete {installer.cache_root}?"): - shutil.rmtree(installer.cache_root) - - -def opkg_install( - installer: RobotpyInstaller, - force_reinstall: bool, - robot: str, - ignore_image_version: bool, - packages: typing.Sequence[pathlib.Path], -): - """ - Installs opkg package on RoboRIO - """ - - installer.log_startup() - for package in packages: - if package.parent != installer.opkg_cache: - raise ValueError("internal error") - if not package.exists(): - raise ClickException(f"{package.name} has not been downloaded yet") + def opkg_install( + self, + force_reinstall: bool, + packages: typing.Sequence[pathlib.Path], + ): + """ + Installs opkg package on RoboRIO + """ - # Write out the install script - # -> we use a script because opkg doesn't have a good mechanism - # to only install a package if it's not already installed - opkg_files = [] - - with installer.get_ssh(robot) as ssh: - cache = installer.start_cache(ssh) + for package in packages: + if package.parent != self.opkg_cache: + raise ValueError("internal error") + if not package.exists(): + raise InstallerException(f"{package.name} has not been downloaded yet") - roborio_checks(ssh, ignore_image_version) + # Write out the install script + # -> we use a script because opkg doesn't have a good mechanism + # to only install a package if it's not already installed + opkg_files = [] opkg_script = inspect.cleandoc( """ @@ -312,7 +155,7 @@ def opkg_install( opkg_script_bit = inspect.cleandoc( f""" if ! opkg list-installed | grep -F "%(name)s - %(version)s"; then - PACKAGES+=("http://localhost:{cache.port}/opkg_cache/%(fname)s") + PACKAGES+=("http://localhost:{self.cache_server.port}/opkg_cache/%(fname)s") DO_INSTALL=1 else echo "%(name)s already installed" @@ -354,242 +197,270 @@ def opkg_install( with catch_ssh_error("creating opkg install script"): # write to /tmp so that it doesn't persist - ssh.exec_cmd( + self.ssh.exec_cmd( f"echo '{opkg_script}' > /tmp/install_opkg.sh", check=True, ) with catch_ssh_error("installing selected packages"): - ssh.exec_cmd("bash /tmp/install_opkg.sh", check=True, print_output=True) + self.ssh.exec_cmd( + "bash /tmp/install_opkg.sh", check=True, print_output=True + ) try: - ssh.exec_cmd("rm /tmp/install_opkg.sh") + self.ssh.exec_cmd("rm /tmp/install_opkg.sh") except SshExecError: pass - show_disk_space(ssh) + def show_disk_space( + self, + ) -> typing.Tuple[str, str, str]: + # + # Free space check.. maybe in the future we'll use this to not accidentally + # fill the user's disk, but it'd be annoying to figure out + # + with catch_ssh_error("checking free space"): + result = self.ssh.check_output("df -h / | tail -n 1") -# -# python installation -# + _, size, used, _, pct, _ = result.strip().split() + logger.info("-> RoboRIO disk usage %s/%s (%s full)", used, size, pct) + return size, used, pct -def _get_python_ipk_path(installer: RobotpyInstaller) -> pathlib.Path: - parts = urlparse(_PYTHON_IPK) - return installer.opkg_cache / pathlib.PurePosixPath(parts.path).name + def ensure_image_version(self, ignore_image_version: bool): + if self._image_version_ok: + return + with catch_ssh_error("retrieving image version"): + result = self.ssh.check_output( + "grep IMAGEVERSION /etc/natinst/share/scs_imagemetadata.ini", + ) -@installer.command() -@option("--use-certifi", is_flag=True, help="Use SSL certificates from certifi") -@pass_obj -def download_python(installer: RobotpyInstaller, use_certifi: bool): - """ - Downloads Python to a folder to be installed - """ - installer.opkg_cache.mkdir(parents=True, exist_ok=True) + roborio_match = re.match( + r'IMAGEVERSION = "(FRC_)?roboRIO_(.*)"', result.strip() + ) + roborio2_match = re.match( + r'IMAGEVERSION = "(FRC_)?roboRIO2_(.*)"', result.strip() + ) - ipk_dst = _get_python_ipk_path(installer) - _urlretrieve(_PYTHON_IPK, ipk_dst, True, _make_ssl_context(use_certifi)) + if roborio_match: + version = roborio_match.group(2) + images = _ROBORIO_IMAGES + name = "RoboRIO" + elif roborio2_match: + version = roborio2_match.group(2) + images = _ROBORIO2_IMAGES + name = "RoboRIO 2" + else: + version = "" + images = [ + f"({_ROBORIO_IMAGES[-1]} | {_ROBORIO2_IMAGES[-1]})", + ] + name = "RoboRIO (1 | 2)" + + logger.info(f"-> {name} image version: {version}") + + if not ignore_image_version and version not in images: + raise InstallerException( + f"{name} image {images[-1]} is required!\n" + "\n" + "See https://docs.wpilib.org/en/stable/docs/zero-to-robot/step-3/imaging-your-roborio.html\n" + "for information about upgrading the RoboRIO image.\n" + "\n" + "Use --ignore-image-version to install anyways" + ) + self._image_version_ok = True -@installer.command() -@_common_ssh_options -@pass_obj -def install_python( - installer: RobotpyInstaller, - robot: str, - ignore_image_version: bool, -): - """ - Installs Python on a RoboRIO. + def ensure_robot_pip(self): + if self._robot_pip_ok: + return - Requires download-python to be executed first. - """ - ipk_dst = _get_python_ipk_path(installer) - opkg_install(installer, False, robot, ignore_image_version, [ipk_dst]) + # + # Ensure that pip is installed + # + with catch_ssh_error("checking for pip3"): + if self.ssh.exec_cmd("[ -x /usr/local/bin/pip3 ]").returncode != 0: + raise InstallerException( + inspect.cleandoc( + """ + pip3 not found on RoboRIO, did you install python? -@installer.command() -@_common_ssh_options -@pass_obj -def uninstall_python( - installer: RobotpyInstaller, - robot: str, - ignore_image_version: bool, -): - """Uninstall Python from a RoboRIO""" - installer.log_startup() + Use the 'download-python' and 'install-python' commands first! + """ + ) + ) - with installer.get_ssh(robot) as ssh: - roborio_checks(ssh, ignore_image_version) + # Use pip stub to override the wheel platform on roborio + with catch_ssh_error("copying pip stub"): + from . import _pipstub - with catch_ssh_error("removing packages"): - ssh.exec_cmd( - f"opkg remove {_ROBOTPY_PYTHON_VERSION}", check=True, print_output=True + stub_fp = io.BytesIO() + stub_fp.write(b"#!/usr/local/bin/python3\n\n") + stub_fp.write(inspect.getsource(_pipstub).encode("utf-8")) + stub_fp.seek(0) + + self.ssh.sftp_fp(stub_fp, _PIP_STUB_PATH) + self.ssh.exec_cmd(f"chmod +x {_PIP_STUB_PATH}", check=True) + + self._robot_pip_ok = True + + # + # Python installation + # + + @property + def _python_ipk_path(self) -> pathlib.Path: + parts = urlparse(_PYTHON_IPK) + return self.opkg_cache / pathlib.PurePosixPath(parts.path).name + + def download_python(self, use_certifi: bool): + self.opkg_cache.mkdir(parents=True, exist_ok=True) + + ipk_dst = self._python_ipk_path + _urlretrieve(_PYTHON_IPK, ipk_dst, True, _make_ssl_context(use_certifi)) + + def install_python(self): + """ + Installs Python on a RoboRIO. + + Requires download-python to be executed first. + """ + ipk_dst = self._python_ipk_path + self.opkg_install(False, [ipk_dst]) + + def uninstall_python( + self, + ): + with catch_ssh_error("removing python"): + self.ssh.exec_cmd( + f"opkg remove {_ROBOTPY_PYTHON_VERSION}", + check=True, + print_output=True, ) - show_disk_space(ssh) - - -# -# Python package management -# - - -def _pip_options(f): - f = option( - "--force-reinstall", - is_flag=True, - help="When upgrading, reinstall all packages even if they are already up-to-date.", - )(f) - f = option( - "--ignore-installed", - "-I", - is_flag=True, - help="Ignore the installed packages (reinstalling instead)", - )(f) - f = option("--no-deps", is_flag=True, help="Don't install package dependencies")(f) - f = option( - "--pre", - is_flag=True, - default=_IS_BETA, - help="Include pre-release and development versions", - )(f) - f = option( - "--requirements", - "-r", - multiple=True, - type=click.Path(exists=True), - default=[], - help="Install from the given requirements file. This option can be used multiple times.", - )(f) - - f = argument("packages", nargs=-1)(f) - return f - - -def _extend_pip_args( - pip_args: typing.List[str], - cache: typing.Optional[CacheServer], - force_reinstall: bool, - ignore_installed: bool, - no_deps: bool, - pre: bool, - requirements: typing.Sequence[str], -): - if pre: - pip_args.append("--pre") - if force_reinstall: - pip_args.append("--force-reinstall") - if ignore_installed: - pip_args.append("--ignore-installed") - if no_deps: - pip_args.append("--no-deps") - - for req in requirements: - if cache: - fname = f"/requirements/{basename(req)}" - cache.add_mapping(fname, req) - pip_args.extend(["-r", f"http://localhost:{cache.port}{fname}"]) - else: - pip_args.extend(["-r", req]) + # + # pip packages + # + def _extend_pip_args( + self, + pip_args: typing.List[str], + cache: typing.Optional[CacheServer], + force_reinstall: bool, + ignore_installed: bool, + no_deps: bool, + pre: bool, + requirements: typing.Iterable[str], + ): + if pre: + pip_args.append("--pre") + if force_reinstall: + pip_args.append("--force-reinstall") + if ignore_installed: + pip_args.append("--ignore-installed") + if no_deps: + pip_args.append("--no-deps") + + for req in requirements: + if cache: + fname = f"/requirements/{basename(req)}" + cache.add_mapping(fname, req) + pip_args.extend(["-r", f"http://localhost:{cache.port}{fname}"]) + else: + pip_args.extend(["-r", req]) + + def pip_download( + self, + no_deps: bool, + pre: bool, + requirements: typing.Iterable[str], + packages: typing.Iterable[str], + ): + """ + Specify Python package(s) to download, and store them in the cache. + + You must be connected to the internet for this to work. + """ + + if not requirements and not packages: + raise InstallerException( + "You must give at least one requirement to download" + ) -@installer.command() -@_pip_options -@pass_obj -def download( - installer: RobotpyInstaller, - force_reinstall: bool, - ignore_installed: bool, - no_deps: bool, - pre: bool, - requirements: typing.Tuple[str], - packages: typing.Tuple[str], -): - """ - Specify Python package(s) to download, and store them in the cache. + try: + import pip # type: ignore + except ImportError: + raise InstallerException( + "ERROR: pip must be installed to download python packages" + ) + + self.pip_cache.mkdir(parents=True, exist_ok=True) - You must be connected to the internet for this to work. - """ + pip_args = [ + "--no-cache-dir", + "--disable-pip-version-check", + "download", + "--extra-index-url", + _ROBORIO_WHEELS, + "--only-binary", + ":all:", + "--platform", + _ROBOTPY_PYTHON_PLATFORM, + "--python-version", + _ROBOTPY_PYTHON_VERSION_NUM, + "--implementation", + "cp", + "--abi", + f"cp{_ROBOTPY_PYTHON_VERSION_NUM}", + "-d", + str(self.pip_cache), + ] - installer.log_startup() + self._extend_pip_args( + pip_args, + None, + False, + False, + no_deps, + pre, + requirements, + ) - if not requirements and not packages: - raise ClickException("You must give at least one requirement to download") + pip_args.extend(packages) + pip_args = [sys.executable, "-m", "robotpy_installer._pipstub"] + pip_args + + logger.debug("Using pip to download: %s", pip_args) + + retval = subprocess.call(pip_args) + if retval != 0: + raise InstallerException("pip download failed") + + def pip_install( + self, + force_reinstall: bool, + ignore_installed: bool, + no_deps: bool, + pre: bool, + requirements: typing.Sequence[str], + packages: typing.Sequence[str], + ): + """ + Installs Python package(s) on a RoboRIO. + + The package must already been downloaded with the 'download' command first. + """ + + self.ensure_robot_pip() + + if len(requirements) == 0 and len(packages) == 0: + raise InstallerException( + "You must give at least one requirement to install" + ) - try: - import pip # type: ignore - except ImportError: - raise ClickException("ERROR: pip must be installed to download python packages") - - installer.pip_cache.mkdir(parents=True, exist_ok=True) - - pip_args = [ - "--no-cache-dir", - "--disable-pip-version-check", - "download", - "--extra-index-url", - _ROBORIO_WHEELS, - "--only-binary", - ":all:", - "--platform", - _ROBOTPY_PYTHON_PLATFORM, - "--python-version", - _ROBOTPY_PYTHON_VERSION_NUM, - "--implementation", - "cp", - "--abi", - f"cp{_ROBOTPY_PYTHON_VERSION_NUM}", - "-d", - str(installer.pip_cache), - ] - - _extend_pip_args( - pip_args, None, force_reinstall, ignore_installed, no_deps, pre, requirements - ) - - pip_args.extend(packages) - pip_args = [sys.executable, "-m", "robotpy_installer._pipstub"] + pip_args - - logger.debug("Using pip to download: %s", pip_args) - - retval = subprocess.call(pip_args) - if retval != 0: - raise ClickException("pip download failed") - - -@installer.command(name="install") -@_pip_options -@_common_ssh_options -@pass_obj -def pip_install( - installer: RobotpyInstaller, - force_reinstall: bool, - ignore_installed: bool, - no_deps: bool, - pre: bool, - requirements: typing.Tuple[str], - packages: typing.Tuple[str], - ignore_image_version: bool, - robot: str, -): - """ - Installs Python package(s) on a RoboRIO. - - The package must already been downloaded with the 'download' command first. - """ - - installer.log_startup() - - if len(requirements) == 0 and len(packages) == 0: - raise ClickException("You must give at least one requirement to install") - - with installer.get_ssh(robot) as ssh: - roborio_checks(ssh, ignore_image_version, pip_check=True) - - cachesvr = installer.start_cache(ssh) + cache_server = self.cache_server pip_args = [ "/home/admin/rpip", @@ -599,15 +470,15 @@ def pip_install( "--no-index", "--root-user-action=ignore", "--find-links", - f"http://localhost:{cachesvr.port}/pip_cache/", + f"http://localhost:{cache_server.port}/pip_cache/", # always add --upgrade, anything in the cache should be installed "--upgrade", "--upgrade-strategy=eager", ] - _extend_pip_args( + self._extend_pip_args( pip_args, - cachesvr, + cache_server, force_reinstall, ignore_installed, no_deps, @@ -618,68 +489,30 @@ def pip_install( pip_args.extend(packages) with catch_ssh_error("installing packages"): - ssh.exec_cmd(" ".join(pip_args), check=True, print_output=True) + self.ssh.exec_cmd(" ".join(pip_args), check=True, print_output=True) # Some of our hacky wheels require this with catch_ssh_error("running ldconfig"): - ssh.exec_cmd("ldconfig") + self.ssh.exec_cmd("ldconfig") - show_disk_space(ssh) - - -@installer.command(name="list") -@_common_ssh_options -@pass_obj -def pip_list( - installer: RobotpyInstaller, - ignore_image_version: bool, - robot: str, -): - """ - Lists Python packages present on RoboRIO - """ - installer.log_startup() - - with installer.get_ssh(robot) as ssh: - roborio_checks(ssh, ignore_image_version, pip_check=True) + def pip_list(self): + self.ensure_robot_pip() with catch_ssh_error("pip3 list"): - ssh.exec_cmd( + self.ssh.exec_cmd( f"{_PIP_STUB_PATH} --no-cache-dir --disable-pip-version-check list", check=True, print_output=True, ) + def pip_uninstall( + self, + packages: typing.Sequence[str], + ): + self.ensure_robot_pip() -@installer.command(name="uninstall") -@_common_ssh_options -@option( - "--requirements", - "-r", - multiple=True, - type=click.Path(exists=True), - default=[], - help="Install from the given requirements file. This option can be used multiple times.", -) -@argument("packages", nargs=-1) -@pass_obj -def pip_uninstall( - installer: RobotpyInstaller, - requirements: typing.Tuple[str], - packages: typing.Tuple[str], - ignore_image_version: bool, - robot: str, -): - """ - Uninstall Python packages from a RoboRIO - """ - installer.log_startup() - - if len(requirements) == 0 and len(packages) == 0: - raise ClickException("You must give at least one requirement to install") - - with installer.get_ssh(robot) as ssh: - roborio_checks(ssh, ignore_image_version, pip_check=True) + if len(packages) == 0: + raise InstallerException("You must give at least one package to uninstall") pip_args = [ _PIP_STUB_PATH, @@ -691,50 +524,26 @@ def pip_uninstall( pip_args.extend(packages) with catch_ssh_error("uninstalling packages"): - ssh.exec_cmd(" ".join(pip_args), check=True, print_output=True) - - -# -# Removed commands -# + self.ssh.exec_cmd(" ".join(pip_args), check=True, print_output=True) -@installer.command(hidden=True) -def download_robotpy(): - raise ClickException( - inspect.cleandoc( - """ - - The download-robotpy command has been removed! The equivalent commands are now: - - robotpy-installer download-python - robotpy-installer download robotpy +def _make_ssl_context(use_certifi: bool): + if not use_certifi: + return None - Run "robotpy-installer --help" for details. - """ + try: + import certifi # type: ignore + except ImportError: + raise InstallerException( + "certifi is not installed, please install it via `pip install certifi`" ) - ) - - -@installer.command(hidden=True) -def install_robotpy(): - raise ClickException( - inspect.cleandoc( - """ - - The install-robotpy command has been removed! The equivalent commands are now: - robotpy-installer install-python - robotpy-installer install robotpy - - Run "robotpy-installer --help" for details. - """ - ) - ) + import ssl + return ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certifi.where()) -# alias for backwards compat -main = installer -if __name__ == "__main__": - installer() +def main(): + print("ERROR: robotpy-installer is now a subcommand of 'robotpy'", file=sys.stderr) + print("- Use 'python -m robotpy installer'", file=sys.stderr) + sys.exit(1) diff --git a/robotpy_installer/sshcontroller.py b/robotpy_installer/sshcontroller.py index a33522d..7f65be4 100644 --- a/robotpy_installer/sshcontroller.py +++ b/robotpy_installer/sshcontroller.py @@ -1,10 +1,9 @@ -import configparser import io import logging import re import os from os.path import exists, join, expanduser, split as splitpath -from pathlib import PurePath, PurePosixPath +from pathlib import Path, PurePath, PurePosixPath import socket import sys import typing @@ -12,9 +11,11 @@ import paramiko -from robotpy_installer.errors import SshExecError, Error -from robotpy_installer.robotfinder import RobotFinder -from robotpy_installer.utils import _resolve_addr +from .errors import SshExecError, Error +from .robotfinder import RobotFinder +from .utils import _resolve_addr + +from . import wpilib_preferences logger = logging.getLogger("robotpy.installer") @@ -160,74 +161,93 @@ def sftp_fp(self, fp, remote_path): def ssh_from_cfg( - cfg_filename: typing.Union[str, os.PathLike], + project_path: Path, + main_file: Path, username: str, password: str, - hostname: typing.Optional[typing.Union[str, int]] = None, + robot_or_team: typing.Union[None, str, int] = None, no_resolve=False, ): - # hostname can be a team number or an ip / hostname - - dirty = True - cfg = configparser.ConfigParser() - cfg.setdefault("auth", {}) - - if exists(cfg_filename): - logger.info("-> using existing config at '%s'", str(cfg_filename)) - cfg.read(cfg_filename) + try: + prefs = wpilib_preferences.load(project_path) dirty = False - - if hostname is not None: + except FileNotFoundError: + prefs = wpilib_preferences.WPILibPreferencesJson() dirty = True - cfg["auth"]["hostname"] = str(hostname) - hostname = cfg["auth"].get("hostname") + if robot_or_team is not None: + if isinstance(robot_or_team, int): + prefs.teamNumber = robot_or_team + else: + prefs.robotHostname = robot_or_team - if not hostname: + if prefs.teamNumber is None and prefs.robotHostname is None: dirty = True print("Robot setup (hit enter for default value):") - while not hostname: - hostname = input("Team number or robot hostname: ") + response = "" + while not response: + response = input("Team number or robot hostname: ") - cfg["auth"]["hostname"] = hostname + try: + prefs.teamNumber = int(response) + except ValueError: + prefs.robotHostname = response if dirty: - with open(cfg_filename, "w") as fp: - cfg.write(fp) + # Only write preferences file if this is a robot project + if main_file.exists(): + prefs.write(project_path) + else: + logger.info( + "-> not saving robot preferences as this isn't a robot project directory" + ) - # see if an ssh alias exists - try: - with open(join(expanduser("~"), ".ssh", "config")) as fp: - hn = hostname.lower() - for line in fp: - if re.match(r"\s*host\s+%s\s*" % hn, line.lower()): - no_resolve = True - break - except Exception: - pass - - # check to see if this is a team number - team = None - try: - team = int(hostname.strip()) - except ValueError: - # check to see if it matches a team hostname - # -> allows legacy hostname configurations to benefit from - # the robot finder - if not no_resolve: - hostmod = hostname.lower().strip() - m = re.search(r"10.(\d+).(\d+).2", hostmod) - if m: - team = int(m.group(1)) * 100 + int(m.group(2)) - else: - m = re.match(r"roborio-(\d+)-frc(?:\.(?:local|lan))?$", hostmod) + team: typing.Optional[int] = prefs.teamNumber + hostname: typing.Optional[str] = prefs.robotHostname + + # Prefer a hostname if specified + if hostname: + # see if an ssh alias exists + try: + with open(join(expanduser("~"), ".ssh", "config")) as fp: + hn = hostname.lower() + for line in fp: + if re.match(r"\s*host\s+%s\s*" % hn, line.lower()): + no_resolve = True + break + except Exception: + pass + + # Attempt to convert it to a team number, which allows users to + # benefit from the robot finder + try: + team = int(hostname.strip()) + except ValueError: + if not no_resolve: + hostmod = hostname.lower().strip() + m = re.search(r"10.(\d+).(\d+).2", hostmod) if m: - team = int(m.group(1)) + team = int(m.group(1)) * 100 + int(m.group(2)) + hostname = None + else: + m = re.match(r"roborio-(\d+)-frc(?:\.(?:local|lan))?$", hostmod) + if m: + team = int(m.group(1)) + hostname = None + else: + hostname = None conn = None - if team: + assert team is not None or hostname is not None + + if hostname is not None: + if no_resolve: + conn_hostname = hostname + else: + conn_hostname = _resolve_addr(hostname) + elif team is not None: logger.info("Finding robot for team %s", team) finder = RobotFinder( ("10.%d.%d.2" % (team // 100, team % 100), False), @@ -244,10 +264,7 @@ def ssh_from_cfg( no_resolve = True conn_hostname, conn = answer else: - conn_hostname = hostname - - if not no_resolve: - conn_hostname = _resolve_addr(hostname) + raise Error("internal logic error") logger.info("Connecting to robot via SSH at %s", conn_hostname) diff --git a/robotpy_installer/utils.py b/robotpy_installer/utils.py index 6942b60..c3c6bfa 100644 --- a/robotpy_installer/utils.py +++ b/robotpy_installer/utils.py @@ -1,3 +1,4 @@ +import functools import contextlib import hashlib import json @@ -8,7 +9,7 @@ import urllib.request from robotpy_installer import __version__ -from robotpy_installer.errors import Error +from .errors import Error logger = logging.getLogger("robotpy.installer") @@ -160,3 +161,15 @@ def yesno(prompt: str) -> bool: a = input(prompt).lower() return a == "y" + + +def handle_cli_error(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except Error as e: + print(f"ERROR:", e, file=sys.stderr) + return False + + return wrapper diff --git a/robotpy_installer/wpilib_preferences.py b/robotpy_installer/wpilib_preferences.py new file mode 100644 index 0000000..da7d22f --- /dev/null +++ b/robotpy_installer/wpilib_preferences.py @@ -0,0 +1,85 @@ +import dataclasses +import logging +import json +import pathlib +import typing + + +logger = logging.getLogger("robotpy.installer") + + +def _wpilib_preferences_json_path(project_path: pathlib.Path): + return project_path / ".wpilib" / "wpilib_preferences.json" + + +@dataclasses.dataclass +class WPILibPreferencesJson: + #: current language + currentLanguage: typing.Optional[str] = None + #: project year + projectYear: typing.Optional[str] = None + #: team number + teamNumber: typing.Optional[int] = None + #: robot hostname -- should never need to specify this + robotHostname: typing.Optional[str] = None + + def write(self, project_path: pathlib.Path): + """ + Writes this wpilib_preferences.json file to disk + + :param project_path: Path to robot project + """ + data = dataclasses.asdict(self) + data = {k: v for k, v in data.items() if v is not None} + + fname = _wpilib_preferences_json_path(project_path) + fname.parent.mkdir(parents=True, exist_ok=True) + + with open(fname, "w") as fp: + json.dump(data, fp) + + logger.info("Settings stored at %s", fname) + + +def load(project_path: pathlib.Path) -> WPILibPreferencesJson: + """ + Reads the project's wpilib_preferences.json from disk. Raises FileNotFoundError + if not present. + + :param project_path: Path to robot project + """ + + wpilib_preferences_json = _wpilib_preferences_json_path(project_path) + + with open(wpilib_preferences_json, "r") as fp: + data = json.load(fp) + + logger.info("Settings loaded from %s", wpilib_preferences_json) + + currentLanguage = data.get("currentLanguage", None) + if currentLanguage is not None: + currentLanguage = str(currentLanguage) + + projectYear = data.get("projectYear", None) + if projectYear is not None: + projectYear = str(projectYear) + + teamNumber = data.get("teamNumber", None) + if teamNumber is not None: + try: + teamNumber = int(teamNumber) + except ValueError: + raise ValueError( + f"{wpilib_preferences_json}: teamNumber must be an integer (got {teamNumber!r})" + ) from None + + robotHostname = data.get("robotHostname", None) + if robotHostname is not None: + robotHostname = str(robotHostname) + + return WPILibPreferencesJson( + currentLanguage, + projectYear, + teamNumber, + robotHostname, + ) diff --git a/setup.cfg b/setup.cfg index 8d45e47..068b879 100644 --- a/setup.cfg +++ b/setup.cfg @@ -24,7 +24,6 @@ zip_safe = False include_package_data = True packages = find: install_requires = - click paramiko pynetconsole~=2.0.2 robotpy-cli~=2024.0b @@ -38,4 +37,5 @@ console_scripts = robotpy = deploy = robotpy_installer.cli_deploy:Deploy deploy-info = robotpy_installer.cli_deploy_info:DeployInfo + installer = robotpy_installer.cli_installer:Installer undeploy = robotpy_installer.cli_undeploy:Undeploy diff --git a/tests/test_basic.py b/tests/test_basic.py index 11da692..1795d67 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -8,7 +8,8 @@ def test_download_basic(): [ sys.executable, "-m", - "robotpy_installer", + "robotpy", + "installer", "download", "-r", str(pathlib.Path(__file__).parent / "sample-requirements.txt"),