Naumenko/clean_annotations (#2)

* feat(commands): initial commit for clean_annotations

* chore(deps): update requirements.txt

* fix(defaults): fix warnings with config.json while testing

* feat(annotations): implement operation for removing orphan annotations

* fix(annotations): make a_suffix Tuple type; make a_source argument setter, if a_source is None - a_source = source_directory

* docs(annotations): add clean-annotations command description to readme

---------

Co-authored-by: Serhii Naumenko <naumenko.s.mail@gmail.com>

Author: SeregaCodit

README.MD

@@ -14,6 +14,7 @@ if you don’t want the command works in a cycle, just don't use "-r" argument.
 - **delete** - delete files that match patterns from source directory
 - **dedup** - find duplicates in source directory that matches a pattern. An image means a duplicate if it's hash has lower 
 Hamming distance with comparing image hash than threshold value. The threshold value setups in percentage and must be in range [0, 100]. Pay attention to core_size parameter: the lower value makes details at photo less important, and the higher value makes details mach important while comparing information at images. It’s implemented only dHash comparing method for now.
+- **clean-annotations** - find annotation files in directory that doesn't have corresponding files
 ## How to use:
 clone git repository:
 

config.json

@@ -14,5 +14,7 @@
   "core_size": 16,
   "n_jobs":  20,
   "cache_file_path": "./cache",
-  "cache_name": null
+  "cache_name": null,
+  "a_suffix": [".xml"],
+  "a_source": null
 }

const_utils/arguments.py

@@ -28,3 +28,5 @@ class Arguments:
     core_size: str = "--core_size"
     n_jobs: str = "--n_jobs"
     cache_name: str = "--cache_name"
+    a_suffix: str = "--a_suffix"
+    a_source: str = "--a_source"

const_utils/commands.py

@@ -6,4 +6,5 @@ class Commands:
     move: str = "move"
     slice: str = "slice"
     delete: str = "delete"
-    dedup: str = "dedup"
+    dedup: str = "dedup"
+    clean_annotations: str = "clean-annotations"

const_utils/default_values.py

@@ -17,8 +17,9 @@ class AppSettings(BaseSettings):
 
     model_config = SettingsConfigDict(
         env_prefix="APP_",
-        json_file=Constants.config_file,
-        extra="ignore"
+        # json_file=Constants.config_file,
+        extra="ignore",
+        validate_assignment=True
     )
 
     remove: bool = Field(default=False)
@@ -37,6 +38,8 @@ class AppSettings(BaseSettings):
     n_jobs: int = Field(default=2, ge=1, le=multiprocessing.cpu_count())
     cache_file_path: Path = Field(default=Path("./cache"))
     cache_name: Optional[Path] = Field(default=None)
+    a_suffix: Tuple[str, ...] = Field(default_factory=tuple)
+    a_source: Optional[Path] = Field(default=None)
 
     @field_validator('core_size')
     @classmethod
@@ -45,13 +48,15 @@ def check_power_of_two(cls, value: int) -> int:
             raise ValueError(f"core_size must be a power of 2 (e.g., 8, 16, 32, 64...), got {value}")
         return value
 
-    @field_validator('log_path', 'cache_file_path', mode='before')
+    @field_validator("log_path", "cache_file_path", "a_source", mode='before')
     @classmethod
     def ensure_path(cls, value: Union[str, Path]) -> Path:
         if isinstance(value, str):
             return Path(value)
         return value
 
+
+
     @classmethod
     def load_config(cls, config_path: Path = Constants.config_file) -> "AppSettings":
 

const_utils/parser_help.py

@@ -29,4 +29,7 @@ class HelpStrings:
     )
     n_jobs: str = "A count of workers for CPU Bound tasks like a hashmap building"
     cache_name: str = ("A cache file name. If you don't set this parameter cache name will be generated automatically "
-                       "with next signature: <cache_{path_hash}_d{folder_name}{hash_type}s{core_size}.pkl>")
+                       "with next signature: <cache_{path_hash}_d{folder_name}{hash_type}s{core_size}.pkl>")
+    a_suffix: str = "A suffix pattern for annotations"
+    a_source: str = ("A source directory to annotations. If None - that means annotations are in the same folder with"
+                     " images")

fileManager.py

@@ -10,7 +10,7 @@
 from file_operations.delete import DeleteOperation
 from file_operations.move import MoveOperation
 from file_operations.slice import SliceOperation
-from logger.log_level_mapping import LevelMapping
+from file_operations.clean_annotations import CleanAnnotationsOperation
 
 
 class FileManager:
@@ -22,7 +22,8 @@ def __init__(self):
             Commands.move: MoveOperation,
             Commands.slice: SliceOperation,
             Commands.delete: DeleteOperation,
-            Commands.dedup: DedupOperation
+            Commands.dedup: DedupOperation,
+            Commands.clean_annotations: CleanAnnotationsOperation
         }
         self.settings = AppSettings.load_config(Constants.config_file)
         self._setup_commands()

file_operations/clean_annotations.py

@@ -0,0 +1,73 @@
+import argparse
+from pathlib import Path
+from typing import Union
+
+from const_utils.arguments import Arguments
+from const_utils.default_values import AppSettings
+from const_utils.parser_help import HelpStrings
+from file_operations.file_operation import FileOperation
+from file_operations.file_remover import FileRemoverMixin
+
+
+
+class CleanAnnotationsOperation(FileOperation, FileRemoverMixin):
+    def __init__(self, **kwargs):
+        """
+        Cleans orphan annotations from same or different paths with images.
+        Unique args:
+        a_source: Path - a path to annotations directory. If None - will be set as source_directory value
+        a_suffix: Tuple[str, ...] - Pattern for annotations file suffix
+        """
+        super().__init__(**kwargs)
+        self.a_source = self.settings.a_source
+
+
+    @staticmethod
+    def add_arguments(settings: AppSettings, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument(
+            Arguments.a_suffix,
+            nargs="+",
+            help=HelpStrings.a_suffix,
+            default=settings.a_suffix,
+        )
+        parser.add_argument(
+            Arguments.a_source,
+            help=HelpStrings.a_source,
+            default=settings.a_source,
+        )
+
+
+    def do_task(self) -> None:
+        self.logger.info(f"Checking for orphan annotations in {self.settings.a_source}")
+        annotation_paths = self.get_files(
+            source_directory=self.a_source,
+            pattern=self.settings.a_suffix
+        )
+
+        image_stems = set(image.stem for image in self.files_for_task)
+
+        orphans_removed = 0
+        for a_path in annotation_paths:
+            if a_path.stem not in image_stems:
+                if self._remove_file(a_path):
+                    orphans_removed += 1
+                    self.logger.info(f"Removed {a_path.stem}")
+
+        self.logger.info(f"Removed {orphans_removed} orphan annotations")
+
+    @property
+    def a_source(self) -> Path:
+        return self._a_source
+
+    @a_source.setter
+    def a_source(self, value: Union[Path, str, None]) -> None:
+        """setter for a_source, it might be set to Path type, or rise Type error """
+        if isinstance(value, Path):
+            self._a_source = value
+        elif isinstance(value, str):
+            self._a_source = Path(value)
+        elif value is None:
+            self._a_source = self.source_directory
+        else:
+            self.logger.error(f"Invalid value for a_source: {value}")
+            raise TypeError(f"Invalid value for a_source, can be Union[Path, str, None], got {type(value)}")

file_operations/file_operation.py

@@ -37,16 +37,17 @@ def __init__(self, settings: AppSettings, **kwargs):
         self.logger.info(f"Started with parameters: {kwargs}")
 
 
-    def get_files(self) -> None:
+    def get_files(self, source_directory: Path, pattern: Union[Tuple[str], Tuple[str, ...]]) -> Tuple[Path]:
         """Get files from source directory that match a set of patterns"""
         files = set()
 
-        for p in self.pattern:
-            current_pattern_files = self.source_directory.glob(f"*{p}*")
+        for p in pattern:
+            current_pattern_files = source_directory.glob(f"*{p}*")
             files.update(current_pattern_files)
 
-        self.files_for_task = tuple(files)
-        self.logger.debug(f"Total files_for_task: {len(self.files_for_task)}")
+        files_for_task = tuple(files)
+        self.logger.debug(f"Total files_for_task: {len(files_for_task)}")
+        return files_for_task
 
     def check_source_directory(self) -> None:
         """Check if source directory is valid"""
@@ -67,7 +68,7 @@ def run(self) -> None:
         self.check_directories()
         while True:
             try:
-                self.get_files()
+                self.files_for_task = self.get_files(source_directory=self.source_directory, pattern=self.pattern)
 
                 if len(self.files_for_task) == 0 and self.repeat:
                     self.logger.info(f"No files found for task'{self.pattern}'. Wait for {self.sleep} seconds...")

file_operations/file_remover.py

@@ -19,12 +19,14 @@ def _remove_all(self, filepaths: Union[List[Path], Tuple[Path], Path]) -> None:
             raise TypeError(f'filepaths should be a list or a tuple or a Path, not {type(filepaths)}')
 
 
-    def _remove_file(self: LoggerProtocol, path: Path) -> None:
-        """deletes one received file"""
+    def _remove_file(self: LoggerProtocol, path: Path) -> bool:
+        """deletes one received file. Returns True or False if file was successfully removed or not"""
         if not path.is_file():
             self.logger.warning(f"{path} is not a file")
         try:
             path.unlink(missing_ok=True)
             self.logger.info(f"{path} removed")
+            return True
         except FileNotFoundError:
             self.logger.warning(f"{path} file not exists, skipping")
+            return False