Improve CLI upload / download experience (#82)

* Repeat dataset selection upon failure

* Flag to include hidden files in upload

* Select files to upload from list or glob

* Let the user try again if no files were selected

* Strip the data/ prefix when downloading files

Author: sminot

cirro/cli/controller.py

@@ -56,13 +56,15 @@ def run_ingest(input_params: UploadArguments, interactive=False):
         return
 
     if interactive:
-        input_params = gather_upload_arguments(input_params, projects, processes)
-        files = input_params['files']
+        input_params, files = gather_upload_arguments(input_params, projects, processes)
+        directory = input_params['data_directory']
     else:
-        files = get_files_in_directory(input_params['data_directory'])
-        if len(files) == 0:
-            raise RuntimeWarning("No files to upload, exiting")
-    directory = input_params['data_directory']
+        directory = input_params['data_directory']
+        files = get_files_in_directory(directory)
+
+    if len(files) == 0:
+        raise RuntimeWarning("No files to upload, exiting")
+
     process = get_item_from_name_or_id(processes, input_params['process'])
     cirro.process.check_dataset_files(files, process.id, directory)
 

cirro/cli/interactive/download_args.py

@@ -6,7 +6,7 @@
 from cirro.api.models.file import File
 from cirro.api.models.project import Project
 from cirro.cli.interactive.common_args import ask_project
-from cirro.cli.interactive.utils import prompt_wrapper, InputError
+from cirro.cli.interactive.utils import ask, prompt_wrapper, InputError
 from cirro.cli.models import DownloadArguments
 from cirro.utils import format_date
 
@@ -32,7 +32,16 @@ def ask_dataset(datasets: List[Dataset], input_value: str) -> str:
     for dataset in datasets:
         if f'{dataset.name} - {dataset.id}' == choice:
             return dataset.id
-    raise InputError("User must select a dataset to download")
+
+    # The user has made a selection which does not match
+    # any of the options available.
+    # This is most likely because there was a typo
+    if ask(
+        'confirm',
+        'The selection does match an option available - try again?'
+    ):
+        return ask_dataset(datasets, input_value)
+    raise InputError("Exiting - no dataset selected")
 
 
 def ask_dataset_files(files: List[File]) -> List[File]:
@@ -61,47 +70,61 @@ def ask_dataset_files(files: List[File]) -> List[File]:
         return ask_dataset_files_glob(files)
 
 
+def strip_prefix(fp: str, prefix: str):
+    assert fp.startswith(prefix), f"Expected {fp} to start with {prefix}"
+    return fp[len(prefix):]
+
+
 def ask_dataset_files_list(files: List[File]) -> List[File]:
     answers = prompt_wrapper({
         'type': 'checkbox',
         'name': 'files',
         'message': 'Select the files to download',
         'choices': [
-            file.relative_path
+            strip_prefix(file.relative_path, "data/")
             for file in files
         ]
     })
 
-    return [
+    selected_files = [
         file
         for file in files
-        if file.relative_path in set(answers['files'])
+        if strip_prefix(file.relative_path, "data/") in set(answers['files'])
     ]
 
+    if len(selected_files) == 0:
+        if ask(
+            "confirm",
+            "No files were selected - try again?"
+        ):
+            return ask_dataset_files_list(files)
+        else:
+            raise RuntimeWarning("No files selected")
+    else:
+        return selected_files
+
 
 def ask_dataset_files_glob(files: List[File]) -> List[File]:
 
-    selected_files = ask_dataset_files_glob_single(files)
-    answers = prompt_wrapper({
-        'type': 'confirm',
-        'name': 'confirm',
-        'message': f'Number of files selected: {len(selected_files):} / {len(files):,}'
-    })
-    while not answers['confirm']:
+    confirmed = False
+    while not confirmed:
         selected_files = ask_dataset_files_glob_single(files)
-        answers = prompt_wrapper({
-            'type': 'confirm',
-            'name': 'confirm',
-            'message': f'Number of files selected: {len(selected_files):} / {len(files):,}'
-        })
+        confirmed = ask(
+            "confirm",
+            f'Number of files selected: {len(selected_files):} / {len(files):,}'
+        )
+
+    if len(selected_files) == 0:
+        raise RuntimeWarning("No files selected")
+
     return selected_files
 
 
 def ask_dataset_files_glob_single(files: List[File]) -> List[File]:
 
     print("All Files:")
     for file in files:
-        print(f" - {file.relative_path}")
+        print(f" - {strip_prefix(file.relative_path, 'data/')}")
 
     answers = prompt_wrapper({
         'type': 'text',
@@ -113,12 +136,12 @@ def ask_dataset_files_glob_single(files: List[File]) -> List[File]:
     selected_files = [
         file
         for file in files
-        if fnmatch(file.relative_path, answers['glob'])
+        if fnmatch(strip_prefix(file.relative_path, "data/"), answers['glob'])
     ]
 
     print("Selected Files:")
     for file in selected_files:
-        print(f" - {file.relative_path}")
+        print(f" - {strip_prefix(file.relative_path, 'data/')}")
 
     return selected_files
 

cirro/cli/interactive/upload_args.py

@@ -5,14 +5,53 @@
 
 from prompt_toolkit.shortcuts import CompleteStyle
 from prompt_toolkit.validation import Validator, ValidationError
-from questionary import Choice
 
 from cirro.api.models.process import Process
 from cirro.api.models.project import Project
 from cirro.cli.interactive.common_args import ask_project
-from cirro.cli.interactive.utils import ask
+from cirro.cli.interactive.utils import ask, prompt_wrapper
 from cirro.cli.models import UploadArguments
-from cirro.file_utils import get_directory_stats, get_files_in_directory
+from cirro.file_utils import get_files_in_directory, get_files_stats
+
+
+def ask_data_directory(input_value: str) -> str:
+    directory_prompt = {
+        'type': 'path',
+        'name': 'data_directory',
+        'message': 'Enter the full path of the data directory',
+        'validate': DataDirectoryValidator,
+        'default': input_value or '',
+        'complete_style': CompleteStyle.READLINE_LIKE,
+        'only_directories': True
+    }
+
+    answers = prompt_wrapper(directory_prompt)
+    return answers['data_directory']
+
+
+def ask_name(input_value: str) -> str:
+    name_prompt = {
+        'type': 'input',
+        'name': 'name',
+        'message': 'What is the name of this dataset?',
+        'validate': lambda val: len(val.strip()) > 0 or 'Please enter a name',
+        'default': input_value or ''
+    }
+
+    answers = prompt_wrapper(name_prompt)
+    return answers['name']
+
+
+def ask_description(input_value: str) -> str:
+    description_prompt = {
+        'type': 'input',
+        'name': 'description',
+        'message': 'Enter a description of the dataset (optional)',
+        'default': input_value or ''
+    }
+
+    answers = prompt_wrapper(description_prompt)
+    return answers['description']
 
 
 class DataDirectoryValidator(Validator):
@@ -25,20 +64,22 @@ def validate(self, document):
             )
 
 
-def confirm_data_directory(directory: str, files: List[str]):
-    stats = get_directory_stats(directory, files)
-    is_accepted = ask(
-        'confirm',
-        f'Please confirm that you wish to upload {stats["numberOfFiles"]} files ({stats["sizeFriendly"]})',
-        default=True
-    )
+def confirm_data_files(data_directory: str, files: List[str]):
+    stats = get_files_stats([
+        Path(data_directory) / file
+        for file in files
+    ])
 
-    if not is_accepted:
+    if not ask(
+        "confirm",
+        f'Please confirm that you wish to upload {stats["numberOfFiles"]} files ({stats["sizeFriendly"]})'
+    ):
         sys.exit(1)
 
 
 def ask_process(processes: List[Process], input_value: str) -> str:
     process_names = [process.name for process in processes]
+    process_names.sort()
     return ask(
         'select',
         'What type of files?',
@@ -47,61 +88,114 @@ def ask_process(processes: List[Process], input_value: str) -> str:
     )
 
 
-def gather_upload_arguments(input_params: UploadArguments, projects: List[Project], processes: List[Process]):
-    input_params['project'] = ask_project(projects, input_params.get('project'))
+def ask_include_hidden() -> bool:
+    return prompt_wrapper({
+        'type': 'confirm',
+        'message': "Include hidden files (expert: e.g. Zarr)",
+        'name': 'include_hidden',
+        'default': False
+    })['include_hidden']
+
+
+def ask_files_in_directory(data_directory) -> List[str]:
+
+    # Ask whether hidden files should be included
+    include_hidden = ask_include_hidden()
 
-    input_params['data_directory'] = ask(
-        'path',
-        'Enter the full path of the data directory',
-        required=True,
-        validate=DataDirectoryValidator,
-        default=input_params.get('data_directory') or '',
-        complete_style=CompleteStyle.READLINE_LIKE,
-        only_directories=True
+    # Get the list of all files in the directory
+    # (relative to the data_directory)
+    files = get_files_in_directory(
+        data_directory,
+        include_hidden=include_hidden
     )
 
-    upload_method = ask(
+    choices = [
+        "Upload all files",
+        "Select files from a list",
+        "Select files with a naming pattern (glob)"
+    ]
+
+    choice = ask(
         'select',
-        'What files would you like to upload?',
-        choices=[
-            Choice('Upload all files in directory', 'all'),
-            Choice('Choose files from a list', 'select'),
-            Choice('Specify a glob pattern', 'glob'),
-        ]
+        'Which files would you like to upload from this dataset?',
+        choices=choices
     )
-    input_params['files'] = get_files_in_directory(input_params['data_directory'])
-    if upload_method == 'select':
-        input_params['files'] = ask(
-            'checkbox',
-            'Select the files you wish to upload',
-            choices=input_params['files']
+
+    if choice == choices[0]:
+        return files
+    elif choice == choices[1]:
+        return ask_dataset_files_list(files)
+    else:
+        return ask_dataset_files_glob(files)
+
+
+def ask_dataset_files_list(files: List[str]) -> List[str]:
+    return prompt_wrapper({
+        'type': 'checkbox',
+        'name': 'files',
+        'message': 'Select the files to upload',
+        'choices': files
+    })['files']
+
+
+def ask_dataset_files_glob(files: List[str]) -> List[str]:
+
+    selected_files = ask_dataset_files_glob_single(files)
+    confirmed = ask(
+        "confirm",
+        f'Number of files selected: {len(selected_files):} / {len(files):,}'
+    )
+    while not confirmed:
+        selected_files = ask_dataset_files_glob_single(files)
+        confirmed = ask(
+            "confirm",
+            f'Number of files selected: {len(selected_files):} / {len(files):,}'
         )
-    elif upload_method == 'glob':
-        matching_files = None
-        while not matching_files:
-            glob_pattern = ask('text', 'Glob pattern:')
-            matching_files = [f for f in input_params['files'] if fnmatch(f, glob_pattern)]
-            if len(matching_files) == 0:
-                print('Glob pattern does not match any files, please specify another')
 
-        input_params['files'] = matching_files
+    if len(selected_files) == 0:
+        raise RuntimeWarning("No files selected")
+
+    return selected_files
+
+
+def ask_dataset_files_glob_single(files: List[str]) -> List[str]:
+
+    print("All Files:")
+    for file in files:
+        print(f" - {file}")
+
+    answers = prompt_wrapper({
+        'type': 'text',
+        'name': 'glob',
+        'message': 'Select files by naming pattern (using the * wildcard)',
+        'default': '*'
+    })
+
+    selected_files = [
+        file
+        for file in files
+        if fnmatch(file, answers['glob'])
+    ]
+
+    print("Selected Files:")
+    for file in selected_files:
+        print(f" - {file}")
 
-    confirm_data_directory(input_params['data_directory'], input_params['files'])
+    return selected_files
+
+
+def gather_upload_arguments(input_params: UploadArguments, projects: List[Project], processes: List[Process]):
+    input_params['project'] = ask_project(projects, input_params.get('project'))
+
+    input_params['data_directory'] = ask_data_directory(input_params.get('data_directory'))
+    files = ask_files_in_directory(input_params['data_directory'])
+
+    confirm_data_files(input_params['data_directory'], files)
 
     input_params['process'] = ask_process(processes, input_params.get('process'))
 
     data_directory_name = Path(input_params['data_directory']).name
     default_name = input_params.get('name') or data_directory_name
-    input_params['name'] = ask(
-        'text',
-        'What is the name of this dataset?',
-        default=default_name,
-        validate=lambda val: len(val.strip()) > 0 or 'Please enter a name'
-    )
-    input_params['description'] = ask(
-        'text',
-        'Enter a description of the dataset (optional)',
-        default=input_params.get('description') or ''
-    )
-
-    return input_params
+    input_params['name'] = ask_name(default_name)
+    input_params['description'] = ask_description(input_params.get('description'))
+    return input_params, files