feat: clean up

.gitignore

@@ -171,6 +171,8 @@ src/testmypixipkg/__pycache__/__init__.cpython-312.pyc
 
 data
 nbia-data
+test-data
 
 
 **/.vscode/
+test_fileset.ipynb

.pre-commit-config.yaml

@@ -12,6 +12,9 @@ repos:
       - id: pretty-format-json
       - id: detect-private-key
       - id: debug-statements
+      # - id: no-commit-to-branch
+      #   always_run: true
+      #   args: ['--branch', 'main']
   - repo: https://github.com/timothycrosley/isort
     rev: 5.13.2
     hooks:
@@ -22,13 +25,9 @@ repos:
       - id: pixi-install
         name: Run pixi install
         entry: bash -c 'pixi install --all && git add pixi.lock'
+        always_run: true
         language: system
         files: .+
-      # - id: pixi-run-tests
-      #   name: Run pixi tests
-      #   entry: bash -c 'pixi run test'
-      #   language: system
-      #   files: ^src/.*\.py$|^pyproject\.toml$
       - id: pixi-run-lint
         name: Run pixi lint
         entry: bash -c 'pixi run lint'

src/pydicomsorter/cli.py

@@ -14,18 +14,22 @@
 click.rich_click.STYLE_COMMANDS_TABLE_COLUMN_WIDTH_RATIO = (1, 2)
 
 click.rich_click.OPTION_GROUPS = {
-    'dicomsort': [
+    "dicomsort": [
         {
-            'name': 'Required options',
-            'options': ['--method'],
+            "name": "Required options",
+            "options": ["--method"],
         },
         {
-            'name': 'Advanced options',
-            'options': ['--dry_run', '--overwrite', '--keep_going'],
+            "name": "Advanced options",
+            "options": [
+                "--overwrite",
+                "--keep-going",
+                "--dry-run",
+            ],
         },
         {
-            'name': 'Basic options',
-            'options': ['--version', '--verbose', '--debug', '--help'],
+            "name": "Basic options",
+            "options": ["--version", "--verbose", "--debug", "--help"],
         },
     ]
 }
@@ -39,15 +43,15 @@ def generate_destination_paths(
     return {k: pathlib.Path(fmt % v).resolve() for k, v in dicom_data.items()}
 
 
-@click.command(context_settings={'help_option_names': ['-h', '--help']})
+@click.command(context_settings={"help_option_names": ["-h", "--help"]})
 @click.option(
-    '--method',
-    '-m',
-    type=click.Choice(['move', 'copy', 'link'], case_sensitive=False),
+    "--method",
+    "-m",
+    type=click.Choice(["move", "copy", "link"], case_sensitive=False),
     required=True,
 )
 @click.argument(
-    'sourcedir',
+    "sourcedir",
     type=click.Path(
         exists=True,
         path_type=pathlib.Path,
@@ -56,41 +60,41 @@ def generate_destination_paths(
     ),
 )
 @click.argument(
-    'destination_dir',
+    "destination_dir",
     type=str,
 )
 @click.option(
-    '-n',
-    '--dry_run',
+    "-o",
+    "--overwrite",
     is_flag=True,
-    help='Do not move or copy files, just print what would be done.',
+    default=False,
+    show_default=True,
+    help="Overwrite files if they already exist.",
 )
 @click.option(
-    '-o',
-    '--overwrite',
+    "-k",
+    "--keep-going",
     is_flag=True,
-    default=False,
-    help='Overwrite files if they already exist.',
-    show_default=True,
+    help="Keep going when an error occurs.",
 )
 @click.option(
-    '-k',
-    '--keep_going',
+    "-n",
+    "--dry-run",
     is_flag=True,
-    help='Keep going when an error occurs.',
+    help="Do not move or copy files, just print what would be done.",
 )
 @click.option(
-    '--verbose',
+    "--verbose",
     is_flag=True,
-    help='Print verbose output.',
+    help="Print verbose output.",
 )
 @click.option(
-    '--debug',
+    "--debug",
     is_flag=True,
-    help='Print debug output.',
+    help="Print debug output.",
 )
 @click.version_option()
-@rich_config(help_config={'style_option': 'bold cyan'})
+@rich_config(help_config={"style_option": "bold cyan"})
 def cli(
     sourcedir: pathlib.Path,
     destination_dir: str,
@@ -102,61 +106,92 @@ def cli(
     debug: bool,
 ) -> None:
     """Main function of the package."""
-    import timeit
-
-    start = timeit.default_timer()
-
-    # run find_dicom_files asynchronously
     files: list[pathlib.Path] = find_dicom_files(source_dir=sourcedir)
-    print(f'Found {len(files)} DICOM files.')
+    print(f"Found {len(files)} DICOM files.")
 
     # # other code here
     try:
         sorter = DICOMSorter(destination_dir).validate_keys()
 
     except ValueError as ve:
-        print(f'Error: {ve}')
+        print(f"Error: {ve}")
         return
-    print(f'Keys to use: {sorter.keys}')
 
     file_list: DICOMFileList = DICOMFileList(files).read_tags(sorter.keys)
 
     if dry_run:
+        print(f"Keys to use: {sorter.keys}")
         file_list.summarize(sorter.keys)
         return
 
     destination_paths = generate_destination_paths(file_list.dicom_data, sorter.format)
 
-    print(destination_paths.__len__())
-
-    execute_method(method, destination_paths)
-
-    print(f'Time: {timeit.default_timer() - start}')
-
+    try:
+        execute_method(
+            method,
+            destination_paths,
+            overwrite,
+            keep_going,
+        )
+    except FileExistsError as fee:
+        print(f"Error: {fee}")
+        return
 
 
-def execute_method(method: str, destination_paths: dict[pathlib.Path, pathlib.Path]) -> None:
+def execute_method(
+    method: str,
+    destination_paths: dict[pathlib.Path, pathlib.Path],
+    overwrite: bool,
+    keep_going: bool,
+) -> None:
     """Execute the method on the destination paths."""
     with progress.Progress(
-        '[progress.description]{task.description}',
+        "[progress.description]{task.description}",
         progress.BarColumn(),
-        '[progress.percentage]{task.percentage:>3.0f}%',
+        "[progress.percentage]{task.percentage:>3.0f}%",
         progress.MofNCompleteColumn(),
-        'Time elapsed:',
+        "Time elapsed:",
         progress.TimeElapsedColumn(),
-        'Time remaining:',
+        "Time remaining:",
         progress.TimeRemainingColumn(compact=True),
         refresh_per_second=10,  # bit slower updates
+        transient=True,
     ) as progress2:
-        task = progress2.add_task('Executing method...', total=len(destination_paths))
+        # make sure that method is one of the allowed values
+        assert method in [
+            "move",
+            "copy",
+            "link",
+        ], "Method must be one of 'move', 'copy', or 'link'."
+
+        match method:
+            case "move":
+                msg = "Moving files..."
+            case "copy":
+                msg = "Copying files..."
+            case "link":
+                msg = "Linking files..."
+            case _:
+                raise ValueError(f"Invalid method: {method}")
+
+        task = progress2.add_task(f"{msg:.<21}", total=len(destination_paths))
+
         for source, destination in destination_paths.items():
+            if destination.exists() and not overwrite:
+                print(f"Destination exists: {destination}")
+                if keep_going:
+                    progress2.update(task, advance=1)
+                    continue
+                else:
+                    raise FileExistsError(f"Destination exists: {destination}")
+
             if not destination.parent.exists():
                 destination.parent.mkdir(parents=True)
-            if method == 'move':
+            if method == "move":
                 source.rename(destination)
-            elif method == 'copy':
+            elif method == "copy":
                 copy(source, destination)
-            elif method == 'link':
+            elif method == "link":
                 destination.symlink_to(source)
             progress2.update(task, advance=1)
     return None

src/pydicomsorter/file_list.py

@@ -6,6 +6,7 @@
 
 from pydicomsorter.io import read_tags
 
+
 class DICOMFileList:
     """A class to handle and manipulate a list of paths to dicom files."""
 
@@ -26,6 +27,7 @@ def read_tags(self, tags: list[str]) -> 'DICOMFileList':
             'Time remaining:',
             progress.TimeRemainingColumn(compact=True),
             refresh_per_second=10,  # bit slower updates
+            transient=True,
         ) as progress2:
             task = progress2.add_task('Reading DICOM tags...', total=len(self.files))
             for file in self.files: