[cli] rewrite porting of script workspaces

rtwcli/rtw_cmds/rtw_cmds/workspace/create_verb.py

@@ -936,9 +936,9 @@ def main(self, *, args):
                 ws_name=create_args.upstream_ws_name,
                 ws_folder=create_args.upstream_ws_abs_path,
                 distro=create_args.ros_distro,
-                ws_docker_support=True if create_args.docker else False,
-                docker_tag=create_args.final_image_name if create_args.docker else "",
-                docker_container_name=create_args.container_name if create_args.docker else "",
+                ws_docker_support=create_args.docker,
+                docker_tag=create_args.final_image_name if create_args.docker else None,
+                docker_container_name=create_args.container_name if create_args.docker else None,
                 standalone=create_args.standalone,
             )
             if not update_workspaces_config(WORKSPACES_PATH, local_upstream_ws):

rtwcli/rtw_cmds/rtw_cmds/workspace/delete_verb.py

@@ -26,6 +26,12 @@
 import questionary
 from rtwcli import logger
 from rtwcli.constants import (
+    RICH_TREE_FST_LVL_COLOR,
+    RICH_TREE_GUIDE_STYLE,
+    RICH_TREE_LABEL_COLOR,
+    RICH_TREE_SND_LVL_COLOR,
+    RICH_TREE_STATUS_COLOR,
+    RICH_TREE_TRD_LVL_COLOR,
     WORKSPACES_PATH,
 )
 from rtwcli.docker_utils import (
@@ -39,6 +45,7 @@
 from rtwcli.workspace_utils import (
     Workspace,
     WorkspacesConfig,
+    get_selected_ws_names_from_user,
     load_workspaces_config_from_yaml_file,
     save_workspaces_config,
     workspace_name_completer,
@@ -123,7 +130,7 @@ def execute(self, workspace: Workspace, config: WorkspacesConfig) -> StepStats:
                 raise
         else:
             self.status = StepStatus.SKIPPED
-            self.message = f"Skipping {self.name.value} (not applicable, .e.g. no Docker support)"
+            self.message = f"Skipping {self.name.value} (not applicable, e.g. no Docker support)"
             logger.info(self.message)
 
         return StepStats(name=self.name, status=self.status, message=self.message)
@@ -134,9 +141,13 @@ def __init__(self):
         super().__init__(StepName.REMOVE_DOCKER_CONTAINER)
 
     def should_run(self, workspace: Workspace) -> bool:
-        return bool(workspace.ws_docker_support and workspace.docker_container_name)
+        return workspace.ws_docker_support
 
     def run(self, workspace: Workspace, config: WorkspacesConfig) -> bool:
+        if not workspace.docker_container_name:
+            self.message = "Docker support is enabled but container name not set."
+            return False
+
         if docker_container_exists(workspace.docker_container_name):
             if docker_stop(workspace.docker_container_name):
                 logger.info(f"Stopped Docker container '{workspace.docker_container_name}'")
@@ -166,9 +177,13 @@ def __init__(self):
         super().__init__(StepName.REMOVE_DOCKER_IMAGE)
 
     def should_run(self, workspace: Workspace) -> bool:
-        return bool(workspace.ws_docker_support and workspace.docker_tag)
+        return workspace.ws_docker_support
 
     def run(self, workspace: Workspace, config: WorkspacesConfig) -> bool:
+        if not workspace.docker_tag:
+            self.message = "Docker support is enabled but Docker tag not set."
+            return False
+
         if is_docker_tag_valid(workspace.docker_tag):
             if remove_docker_image(workspace.docker_tag):
                 self.message = f"Removed Docker image '{workspace.docker_tag}'"
@@ -191,6 +206,10 @@ def should_run(self, workspace: Workspace) -> bool:
         return not workspace.standalone
 
     def run(self, workspace: Workspace, config: WorkspacesConfig) -> bool:
+        if not workspace.ws_folder:
+            self.message = "Workspace is not standalone but workspace folder not set."
+            return False
+
         if not os.path.exists(workspace.ws_folder):
             self.message = (
                 f"Workspace folder '{workspace.ws_folder}' does not exist. Skipping removal."
@@ -219,7 +238,14 @@ def run(self, workspace: Workspace, config: WorkspacesConfig) -> bool:
 
         logger.debug(f"Items to delete: {items_to_delete}")
 
-        item_paths_to_delete = [item_choices_to_delete[item] for item in items_to_delete]
+        if len(items_to_delete) == len(items_in_ws_folder):
+            logger.debug(
+                f"User selected to delete all items in workspace folder '{workspace.ws_folder}', "
+                "removing entire folder"
+            )
+            item_paths_to_delete = [workspace.ws_folder]
+        else:
+            item_paths_to_delete = [item_choices_to_delete[item] for item in items_to_delete]
         for item_path in item_paths_to_delete:
             logger.debug(f"Removing item: {item_path}")
             if os.path.isfile(item_path) or os.path.islink(item_path):
@@ -268,27 +294,6 @@ def get_removal_step(step_name: StepName) -> RemovalStep:
     return step_classes[step_name]()
 
 
-def create_choice_format_string(ws_name_width, distro_width, ws_folder_width, num_spaces=2) -> str:
-    return (
-        f"{{ws_name:<{ws_name_width+num_spaces}}}"
-        f" {{distro:<{distro_width+num_spaces}}}"
-        f" {{ws_folder:<{ws_folder_width+num_spaces}}}"
-        f" {{docker_tag}}"
-    )
-
-
-def get_choice_format_template(choice_format: str) -> str:
-    return (
-        "["
-        + (
-            "".join(c for c in choice_format if c.isalpha() or c in "_ {}")
-            .replace("{", "<")
-            .replace("}", ">")
-        )
-        + "]"
-    )
-
-
 def remove_workspaces(
     config: WorkspacesConfig, workspace_names: List[str]
 ) -> Dict[str, WorkspaceRemovalStats]:
@@ -351,64 +356,22 @@ def print_removal_stats(stats: Dict[str, WorkspaceRemovalStats]):
         1 for ws_stats in stats.values() if ws_stats.status == WorkspaceRemovalStatus.COMPLETED
     )
     tree = Tree(
-        f"[bold green]Workspace Removal Statistics: {completed}/{total}", guide_style="bold"
+        f"{RICH_TREE_LABEL_COLOR}Workspace Removal Statistics: {completed}/{total}",
+        guide_style=RICH_TREE_GUIDE_STYLE,
     )
     for ws_i, (ws_name, ws_stats) in enumerate(stats.items(), start=1):
         ws_node = tree.add(
-            f"[bold blue]{ws_i}. {ws_name} - [bold magenta]{ws_stats.status.name}", highlight=True
+            f"{RICH_TREE_FST_LVL_COLOR}{ws_i}. {ws_name} - {RICH_TREE_STATUS_COLOR}{ws_stats.status.name}",
+            highlight=True,
         )
         for step_i, step in enumerate(ws_stats.steps, start=1):
             step_node = ws_node.add(
-                f"[yellow]{step_i}. {step.name.value}: [bold magenta]{step.status.name}"
+                f"{RICH_TREE_SND_LVL_COLOR}{step_i}. {step.name.value}: {RICH_TREE_STATUS_COLOR}{step.status.name}"
             )
-            step_node.add(f"[cyan]{step.message}")
+            step_node.add(f"{RICH_TREE_TRD_LVL_COLOR}{step.message}")
     rich.print(tree)
 
 
-def get_ws_names_to_delete_from_user(workspaces_config: WorkspacesConfig) -> List[str]:
-    logger.debug("ws selection")
-    ws_name_width = max(len(ws_name) for ws_name in workspaces_config.workspaces.keys())
-    distro_width = max(len(ws.distro) for ws in workspaces_config.workspaces.values())
-    ws_folder_width = max(len(ws.ws_folder) for ws in workspaces_config.workspaces.values())
-    choice_format = create_choice_format_string(ws_name_width, distro_width, ws_folder_width)
-
-    choice_data = {}
-    for ws_name, ws in workspaces_config.workspaces.items():
-        ws_choice_data = choice_format.format(
-            ws_name=ws_name,
-            distro=ws.distro,
-            ws_folder=ws.ws_folder,
-            docker_tag=ws.docker_tag,
-        )
-        choice_data[ws_choice_data] = ws_name
-
-    choice_format_template = get_choice_format_template(choice_format)
-    ws_choices_to_delete = questionary.checkbox(
-        message=f"Select workspaces to delete {choice_format_template}\n",
-        choices=list(choice_data.keys()),
-        style=questionary.Style([("highlighted", "bold")]),
-    ).ask()
-    if not ws_choices_to_delete:  # cancelled by user
-        exit(0)
-
-    logger.debug("ws selection confirmation")
-    workspaces_to_delete_to_confirm = "\n".join(
-        f"{i}. {ws_choice}" for i, ws_choice in enumerate(ws_choices_to_delete, start=1)
-    )
-
-    confirm_delete = questionary.confirm(
-        f"Are you sure you want to delete following workspaces? {choice_format_template}"
-        f"\n{workspaces_to_delete_to_confirm}\n"
-    ).ask()
-    if not confirm_delete:  # cancelled by user
-        exit(0)
-
-    ws_names_to_delete = [
-        choice_data[ws_choice_to_delete] for ws_choice_to_delete in ws_choices_to_delete
-    ]
-    return ws_names_to_delete
-
-
 class DeleteVerb(VerbExtension):
     """Delete an available ROS workspace in the config."""
 
@@ -430,7 +393,11 @@ def main(self, *, args):
             logger.debug(f"Deleting workspace from args: {args.workspace_name}")
             ws_names_to_delete = [args.workspace_name]
         else:
-            ws_names_to_delete = get_ws_names_to_delete_from_user(workspaces_config)
+            ws_names_to_delete = get_selected_ws_names_from_user(
+                workspaces=workspaces_config.workspaces,
+                select_question_msg="Select workspaces to delete",
+                confirm_question_msg="Are you sure you want to delete the selected workspaces?",
+            )
 
         logger.debug(f"Ws names to delete: {ws_names_to_delete}")