SIM RFC: Clean up runtime installer/loader code (UBC-Thunderbots#3580)

Author: nycrat

src/software/thunderscope/binary_context_managers/runtime_installer.py

@@ -1,105 +1,67 @@
 import requests
 from pathlib import Path
-from software.logger.logger import create_logger
-import zipfile
 import tarfile
 import shutil
-from software.thunderscope.constants import RuntimeManagerConstants
 import platform
-
-logger = create_logger(__name__)
+from software.thunderscope.constants import RuntimeManagerConstants
 
 
 class RuntimeInstaller:
     """Delegate class for handling runtime installation and remote interfacing"""
 
     def __init__(self):
-        self.download_urls = []
+        self.runtime_install_targets = {}
 
     def fetch_remote_runtimes(self) -> list[str]:
         """Requests a list of available runtimes from the remote. This is an expensive operation
         and should only be called when necessary.
         :return: A unique list of names for available runtimes
         """
-        url = RuntimeManagerConstants.INSTALL_URL
-        headers = {"Accept": "application/vnd.github+json"}
-
-        response = requests.get(url, headers=headers)
-        logger.warning(response)
-
-        releases = response.json()
+        releases = requests.get(
+            RuntimeManagerConstants.RELEASES_URL,
+            headers={"Accept": "application/vnd.github+json"},
+        ).json()
 
-        PREFIX = RuntimeManagerConstants.DOWNLOAD_PREFIX
+        version_names = []
 
-        # I'm going to assume you are trying to reload the assets so reset the download_urls
-        if len(self.download_urls) != 0:
-            self.download_urls = []
-        download_names = []
-        os_name = platform.system()
+        # Currently the only targets that are supported for each os
+        os_to_target = {"Darwin": "mac-arm64", "Linux": "ubuntu-x86"}
+        target = os_to_target[platform.system()]
 
         for release in releases:
-            tag = release.get("tag_name", "")
-
-            # ✅ Only keep tags that include the OS name
-            if os_name not in tag:
-                continue
+            version = release["tag_name"]
             for asset in release.get("assets", []):
                 url = asset["browser_download_url"]
-                if "fullsystem" not in asset_name:
-                    continue
-                self.download_urls.append(url)
-                trimmed = url.removeprefix(PREFIX)
-                download_names.append(trimmed)
 
-        return download_names[:5]
+                if "unix_full_system" in url and target in url:
+                    version_names.append(version)
+                    self.runtime_install_targets[version] = url
+
+        return version_names[: RuntimeManagerConstants.MAX_RELEASES_FETCHED]
 
     def install_runtime(self, version: str) -> None:
         """Installs the runtime of the specified version or throws an error upon failure.
         Ensures that the runtime is compatible with the current platform
         :param version: Version of the runtime hosted on the remote to install
         """
-        url = RuntimeManagerConstants.INSTALL_URL
-        headers = {"Accept": "application/vnd.github+json"}
-
-        response = requests.get(url, headers=headers)
-        logger.warning(response)
-
-        releases = response.json()
-
-        TARGET_SUFFIX = version
-
-        selected_asset = None
-
-        for download_url in self.download_urls:
-            if download_url.endswith(TARGET_SUFFIX):
-                selected_asset = download_url
-
-        if selected_asset:
-            url = selected_asset
-            filename = Path(url).name
-
-            target_dir = Path(RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH)
-            tmp_path = Path("/tmp") / filename
-
-            with requests.get(url, stream=True) as r:
-                r.raise_for_status()
-                with open(tmp_path, "wb") as f:
-                    for chunk in r.iter_content(chunk_size=8192):
-                        if chunk:
-                            f.write(chunk)
-
-            if filename.endswith((".tar.gz", ".tgz")):
-                with tarfile.open(tmp_path, "r:*") as tar:
-                    tar.extractall(path=target_dir)
-
-            elif filename.endswith(".zip"):
-                with zipfile.ZipFile(tmp_path, "r") as zipf:
-                    zipf.extractall(path=target_dir)
-
-            else:
-                dest = target_dir / filename
-                shutil.copy2(tmp_path, dest)
-                dest.chmod(0o755)  # make executable (common for runtimes)
-
-        if not selected_asset:
-            logger.warning("Can't find binary")
+        url = self.runtime_install_targets[version]
+
+        filename = Path(url).name
+        target_dir = Path(RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH)
+        tmp_dir = Path("/tmp")
+        tmp_path = tmp_dir / filename
+        extracted_binary_name = "unix_full_system"
+
+        with requests.get(url, stream=True) as r:
+            r.raise_for_status()
+            with open(tmp_path, "wb") as f:
+                for chunk in r.iter_content(chunk_size=8192):
+                    if chunk:
+                        f.write(chunk)
+
+        dest = target_dir / f"{extracted_binary_name}_{version}"
+
+        # Our release assets for FullSystem are always tar.gz files
+        with tarfile.open(tmp_path, "r:*") as tar:
+            tar.extractall(tmp_dir)
+            shutil.move(tmp_dir / extracted_binary_name, dest)

src/software/thunderscope/binary_context_managers/runtime_loader.py

@@ -1,138 +1,139 @@
 from tomllib import TOMLDecodeError
 from software.thunderscope.constants import RuntimeManagerConstants
-from dataclasses import dataclass
 import os
 import tomllib
 import logging
 
 
-@dataclass
 class RuntimeConfig:
-    """Data class to store the paths of the two binaries"""
+    """Class to store the names and get paths of the two binaries"""
+
+    def __init__(
+        self,
+        blue_runtime: str | None = None,
+        yellow_runtime: str | None = None,
+    ) -> None:
+        """Create runtime config, replacing invalid runtimes with default FullSystem
+        :param blue_runtime: blue runtime name, None if default
+        :param yellow_runtime: yellow runtime name, None if default
+        """
+        self.blue_runtime = (
+            blue_runtime
+            if blue_runtime
+            and self._is_valid_runtime_path(self._get_runtime_path(blue_runtime))
+            else RuntimeManagerConstants.DEFAULT_BINARY_NAME
+        )
+        self.yellow_runtime = (
+            yellow_runtime
+            if yellow_runtime
+            and self._is_valid_runtime_path(self._get_runtime_path(yellow_runtime))
+            else RuntimeManagerConstants.DEFAULT_BINARY_NAME
+        )
+
+    def get_blue_runtime_path(self) -> str:
+        """Returns the path of the stored yellow runtime
+        :return: the absolute path of the binary as a string, or the relative path of our FullSystem
+        """
+        return self._get_runtime_path(self.blue_runtime)
 
-    chosen_blue_path: str = RuntimeManagerConstants.DEFAULT_BINARY_PATH
-    """Blue runtime path"""
+    def get_yellow_runtime_path(self) -> str:
+        """Returns the path of the stored yellow runtime
+        :return: the absolute path of the binary as a string, or the relative path of our FullSystem
+        """
+        return self._get_runtime_path(self.yellow_runtime)
+
+    def _get_runtime_path(self, selected_runtime: str) -> str:
+        """Gets the absolute path of a binary given its name, or the path of our default FullSystem
+        if the binary is not valid.
+        :param selected_runtime: the name of the selected runtime binary
+        :return: the absolute path of the binary as a string, or the relative path of our FullSystem
+        """
+        file_path = os.path.join(
+            RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH, selected_runtime
+        )
+        # Default to local FullSystem if it is selected or the selected binary isn't a valid runtime
+        if (
+            selected_runtime == RuntimeManagerConstants.DEFAULT_BINARY_NAME
+            or not self._is_valid_runtime_path(file_path)
+        ):
+            return RuntimeManagerConstants.DEFAULT_BINARY_PATH
+        # Remove leading and trailing white space and return
+        return file_path.strip()
 
-    chosen_yellow_path: str = RuntimeManagerConstants.DEFAULT_BINARY_PATH
-    """Yellow runtime path"""
+    def _is_valid_runtime_path(self, runtime_path: str) -> bool:
+        """Returns if the binary exists and if it is an executable.
+        :param runtime_path: the path to check
+        :return: True if it is a valid runtime
+        """
+        return os.path.isfile(runtime_path) and os.access(runtime_path, os.X_OK)
 
 
 class RuntimeLoader:
     """Delegate class for handling local runtimes and managing runtime selection"""
 
     def fetch_installed_runtimes(self) -> list[str]:
-        """Fetches the list of available runtimes, including our FullSystem, from the local disk. Makes the folder
-        in our local disk if it does not exist yet.
-        :return: A list of names for available runtimes, or just a list with our FullSystem if no available runtimes
-        could be found
+        """Fetches the list of installed runtimes from the local disk.
+        Creates the external runtimes directory in our local disk if it does not exist yet.
+        :return: A list of installed runtime names
         """
         runtime_list = [RuntimeManagerConstants.DEFAULT_BINARY_NAME]
 
         if not os.path.isdir(RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH):
             os.mkdir(RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH)
-        # Check for all executable files in the folder, and add its name to the list
+
+        # Check for all executable files in the directory, and add its name to the list
         for file_name in os.listdir(RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH):
             file_path = os.path.join(
                 RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH, file_name
             )
             if os.access(file_path, os.X_OK):
                 runtime_list.append(file_name)
 
-        # Cache external runtimes
         return runtime_list
 
-    def load_existing_runtimes(self, yellow_runtime: str, blue_runtime: str) -> None:
+    def load_selected_runtimes(self, yellow_runtime: str, blue_runtime: str) -> None:
         """Loads the yellow and blue runtimes specified by saving them in the local disk.
-        :param blue_runtime: Unique name of the blue runtime to set
-        :param yellow_runtime: Unique name of the yellow runtime to set
+        :param blue_runtime: name of the blue runtime to set
+        :param yellow_runtime: name of the yellow runtime to set
         """
-        config = RuntimeConfig(
-            self._return_runtime_path(blue_runtime),
-            self._return_runtime_path(yellow_runtime),
+        # Format as TOML
+        selected_runtimes = (
+            f'{RuntimeManagerConstants.RUNTIME_CONFIG_BLUE_KEY} = "{blue_runtime}"\n'
+            f'{RuntimeManagerConstants.RUNTIME_CONFIG_YELLOW_KEY} = "{yellow_runtime}"'
         )
-        self._set_runtime_config(config)
+
+        # create a new config file if it doesn't exist, and write in the format above to it
+        with open(RuntimeManagerConstants.RUNTIME_CONFIG_PATH, "w") as file:
+            file.write(selected_runtimes)
 
     def fetch_runtime_config(self) -> RuntimeConfig:
-        """Fetches the runtime configuration from the local disk. If the blue/yellow configuration is invalid,
-        returns the default runtime configuration for blue/yellow
+        """Fetches the runtime configuration from the local disk, creating it if it doesn't exist.
         :return: Returns the runtime configuration as a RuntimeConfig
         """
-        # Create default FullSystem pair with our FullSystem binaries
-        config = RuntimeConfig()
+        # Create empty config file if doesn't exist yet
+        os.makedirs(
+            os.path.dirname(RuntimeManagerConstants.RUNTIME_CONFIG_PATH), exist_ok=True
+        )
+        open(RuntimeManagerConstants.RUNTIME_CONFIG_PATH, "a").close()
 
         try:
             with open(RuntimeManagerConstants.RUNTIME_CONFIG_PATH, "rb") as file:
                 selected_runtime_dict = tomllib.load(file)
-                # Get the persisted blue path, or replace with the default arrangement if it doesn't exist
-                toml_blue_path = selected_runtime_dict.get(
-                    RuntimeManagerConstants.RUNTIME_CONFIG_BLUE_KEY,
-                    RuntimeManagerConstants.DEFAULT_BINARY_PATH,
-                )
-                if self._is_valid_runtime(toml_blue_path):
-                    config.chosen_blue_path = toml_blue_path
-                # Get the persisted yellow path, or replace with the default arrangement if it doesn't exist
-                toml_yellow_path = selected_runtime_dict.get(
-                    RuntimeManagerConstants.RUNTIME_CONFIG_YELLOW_KEY,
-                    RuntimeManagerConstants.DEFAULT_BINARY_PATH,
+
+                # Get the persisted runtimes
+                config = RuntimeConfig(
+                    selected_runtime_dict.get(
+                        RuntimeManagerConstants.RUNTIME_CONFIG_BLUE_KEY
+                    ),
+                    selected_runtime_dict.get(
+                        RuntimeManagerConstants.RUNTIME_CONFIG_YELLOW_KEY
+                    ),
                 )
-                if self._is_valid_runtime(toml_yellow_path):
-                    config.chosen_yellow_path = toml_yellow_path
-        except (FileNotFoundError, PermissionError, TOMLDecodeError):
+
+                return config
+        except TOMLDecodeError:
             logging.warning(
                 f"Failed to read TOML file at: {RuntimeManagerConstants.RUNTIME_CONFIG_PATH}"
             )
 
-        return config
-
-    def _set_runtime_config(self, config: RuntimeConfig) -> None:
-        """Sets/persists the runtime configuration file on disk and creates the configuration
-        file if it doesn't exist.
-        :param config: The runtime configuration containing
-         - color_runtime : absolute path of external runtime, or
-         - color_runtime : relative path of DEFAULT_BINARY_PATH
-        """
-        blue_path = config.chosen_blue_path
-        yellow_path = config.chosen_yellow_path
-
-        """Format in TOML as:
-        blue_path_to_binary: '<runtime path>'
-        yellow_path_to_binary: '<runtime path>'"""
-
-        selected_runtimes = (
-            f'{RuntimeManagerConstants.RUNTIME_CONFIG_BLUE_KEY} = "{blue_path}"\n'
-            f'{RuntimeManagerConstants.RUNTIME_CONFIG_YELLOW_KEY} = "{yellow_path}"'
-        )
-
-        # create a new config file if it doesn't exist, and write in the format above to it
-        with open(RuntimeManagerConstants.RUNTIME_CONFIG_PATH, "w") as file:
-            file.write(selected_runtimes)
-
-    def _return_runtime_path(self, selected_runtime: str) -> str:
-        """Returns the absolute path of a binary given its name, or the path of our default FullSystem
-        if the binary is not valid.
-        :param selected_runtime: the name of the selected runtime binary
-        :return: the absolute path of the binary as a string, or the relative path of our FullSystem
-        """
-        file_path = os.path.join(
-            RuntimeManagerConstants.EXTERNAL_RUNTIMES_PATH, selected_runtime
-        )
-        # Default to our full system if it is selected or the selected binary isn't a valid runtime
-        if (
-            selected_runtime == RuntimeManagerConstants.DEFAULT_BINARY_NAME
-            or not self._is_valid_runtime(file_path)
-        ):
-            return RuntimeManagerConstants.DEFAULT_BINARY_PATH
-        # Remove leading and trailing white space and return
-        return file_path.strip()
-
-    def _is_valid_runtime(self, runtime_path: str) -> bool:
-        """Returns if the path exists and if it is an executable. Logs a warning if it is not valid.
-        :param runtime_path the path to check
-        :return: whether it is a valid runtime or not
-        """
-        if os.path.isfile(runtime_path) and os.access(runtime_path, os.X_OK):
-            return True
-        logging.warning(
-            f"The runtime retrieved at {runtime_path} is not a valid runtime."
-        )
-        return False
+        return RuntimeConfig()

src/software/thunderscope/binary_context_managers/runtime_manager.py

@@ -21,7 +21,7 @@ def fetch_remote_runtimes(self) -> list[str]:
         return self.runtime_installer.fetch_remote_runtimes()
 
     def install_runtime(self, version: str) -> None:
-        """Installs the runtime of the specified version or throws an error upon failure.
+        """Installs the runtime of the specified version or throws an error upon failure
         :param version: Version of the runtime hosted on the remote to install
         """
         self.runtime_installer.install_runtime(version)
@@ -32,12 +32,12 @@ def fetch_installed_runtimes(self) -> list[str]:
         """
         return self.runtime_loader.fetch_installed_runtimes()
 
-    def load_existing_runtimes(self, yellow_runtime: str, blue_runtime: str) -> None:
-        """Loads the runtimes of the specified name or logs a warning upon failure.
+    def load_selected_runtimes(self, yellow_runtime: str, blue_runtime: str) -> None:
+        """Loads the runtimes into the runtime loader config file on the local disk
         :param blue_runtime: name of the blue runtime to load
         :param yellow_runtime: name of the yellow runtime to load
         """
-        self.runtime_loader.load_existing_runtimes(yellow_runtime, blue_runtime)
+        self.runtime_loader.load_selected_runtimes(yellow_runtime, blue_runtime)
 
     def fetch_runtime_config(self) -> RuntimeConfig:
         """Fetches the runtime configuration from the local disk