Merge pull request #53 from nikstuckenbrock/feature/more-detailed-cache-control

Add more detailed cache control

Author: lmandrelli

src/batch_tamarin/commands/cache.py

@@ -0,0 +1,56 @@
+"""
+Cache command for batch-tamarin.
+
+This module handles cache interaction commands.
+"""
+
+import typer
+
+from ..modules.cache_manager import CacheManager
+from ..utils.system_resources import get_human_readable_volume_size
+
+cache_command = typer.Typer(name="cache", help="Interaction with the cache")
+
+
+@cache_command.command()
+def clear(
+    errors_only: bool = typer.Option(
+        False, "--errors-only", "-e", help="Only clear failed/error tasks"
+    )
+) -> None:
+    """Clear the cache.
+
+    Args:
+        errors_only (bool, optional): Clear only error/failed tasks. Defaults to False.
+    """
+
+    CacheCommand.clear(errors_only=errors_only)
+
+
+class CacheCommand:
+    """Command class for interacting with the batch-tamarin cache."""
+
+    @staticmethod
+    def clear(errors_only: bool = False) -> None:
+        """
+        Clears the cache using the given options.
+
+        Args:
+            errors_only (bool, optional): Only clear errors. Defaults to False.
+        """
+
+        try:
+            cache_manager = CacheManager()
+            stats_before = cache_manager.get_stats()
+            cache_manager.clear_cache(errors_only=errors_only)
+            stats_after = cache_manager.get_stats()
+            entries = stats_before["size"] - stats_after["size"]
+            volume = get_human_readable_volume_size(
+                stats_before["volume"] - stats_after["volume"]
+            )
+
+            print(f"Cleared cache: {entries} entries, {volume}")
+        except Exception as e:
+            print(f"Failed to clear cache: {e}")
+            raise typer.Exit(1)
+        return

src/batch_tamarin/main.py

@@ -3,26 +3,24 @@
 import typer
 
 from . import __author__, __version__
+from .commands.cache import cache_command
 from .commands.check import CheckCommand
 from .commands.init import InitCommand
 from .commands.report import ReportCommand
 from .commands.run import RunCommand
 from .model.tamarin_recipe import SchedulingStrategy
-from .modules.cache_manager import CacheManager
 from .utils.notifications import notification_manager
 
 app = typer.Typer(help="batch-tamarin")
+app.add_typer(cache_command)
 
 
 @app.callback(invoke_without_command=True)
 def main_callback(
     ctx: typer.Context,
     version: bool = typer.Option(
         False, "--version", "-v", help="Show version information"
-    ),
-    rm_cache: bool = typer.Option(
-        False, "--rm-cache", help="Remove all cached results"
-    ),
+    )
 ):
     """
     Batch Tamarin - Protocol verification automation tool.
@@ -45,24 +43,6 @@ def main_callback(
         )
         return
 
-    if rm_cache:
-        try:
-            cache_manager = CacheManager()
-            stats = cache_manager.get_stats()
-            cache_manager.clear_cache()
-            # Format volume in human-readable units
-            volume = stats["volume"]
-            unit = "bytes"
-            for unit in ["bytes", "kB", "MB", "GB"]:
-                if volume < 1024 or unit == "GB":
-                    break
-                volume /= 1024
-            print(f"Cleared cache: {stats['size']} entries, {volume:.2f} {unit}")
-        except Exception as e:
-            print(f"Failed to clear cache: {e}")
-            raise typer.Exit(1)
-        return
-
     if ctx.invoked_subcommand is None:
         print(ctx.get_help())
         raise typer.Exit()

src/batch_tamarin/modules/cache_manager.py

@@ -12,7 +12,7 @@
 
 from diskcache import Cache  # type: ignore
 
-from ..model.executable_task import ExecutableTask, TaskResult
+from ..model.executable_task import ExecutableTask, TaskResult, TaskStatus
 from ..utils.notifications import notification_manager
 
 
@@ -86,9 +86,36 @@ def store_result(self, task: ExecutableTask, result: TaskResult) -> None:
         # Store in cache
         self.cache[key] = cached_data
 
-    def clear_cache(self) -> None:
-        """Clear all cached results."""
-        self.cache.clear()
+    def clear_cache(self, errors_only: bool = False) -> None:
+        """
+        Clear all cached results.
+
+        Args:
+            errors_only (bool, optional): Clear only failed/error tasks. Defaults to False.
+        """
+
+        if not errors_only:
+            self.cache.clear()
+        else:
+            for key in self.cache:
+                value = self.cache.get(key)
+                if value.task_result.status in [
+                    TaskStatus.FAILED,
+                    TaskStatus.SIGNAL_INTERRUPTED,
+                    TaskStatus.MEMORY_LIMIT_EXCEEDED,
+                    TaskStatus.TIMEOUT,
+                ]:
+                    self._delete_cache_entry(key)
+
+    def _delete_cache_entry(self, key: str) -> None:
+        """
+        Deletes a single cache entry by key.
+
+        Args:
+            key (str): The key of the entry to delete
+        """
+
+        self.cache.delete(key=key)
 
     def get_stats(self) -> dict[str, int]:
         """

src/batch_tamarin/utils/notifications.py

@@ -14,7 +14,10 @@
 
 from ..model.batch import Batch, LemmaResult
 from ..model.tamarin_recipe import TamarinRecipe
-from ..utils.system_resources import resolve_resource_value
+from ..utils.system_resources import (
+    get_human_readable_volume_size,
+    resolve_resource_value,
+)
 
 if TYPE_CHECKING:
     from ..model.executable_task import ExecutableTask
@@ -306,11 +309,11 @@ def task_execution_summary(self, batch: Batch) -> None:
         )
         runtime_table.add_row(
             "Total peak memory used",
-            f"{self._format_memory(batch.execution_metadata.total_memory)}",
+            f"{get_human_readable_volume_size(batch.execution_metadata.total_memory, 'MB')}",
         )
         runtime_table.add_row(
             "Max peak memory used",
-            f"{self._format_memory(batch.execution_metadata.max_memory)}",
+            f"{get_human_readable_volume_size(batch.execution_metadata.max_memory, 'MB')}",
         )
         runtime_table.add_row(
             "Freshly executed tasks",
@@ -387,8 +390,8 @@ def task_execution_summary(self, batch: Batch) -> None:
                 duration_display = self._format_duration(
                     rich_executable.task_execution_metadata.exec_duration_monotonic
                 )
-                memory_display = self._format_memory(
-                    rich_executable.task_execution_metadata.peak_memory
+                memory_display = get_human_readable_volume_size(
+                    rich_executable.task_execution_metadata.peak_memory, "MB"
                 )
 
                 cache_display = (
@@ -627,48 +630,6 @@ def _format_duration(self, seconds: float) -> str:
             minutes = int((seconds % 3600) // 60)
             return f"{hours}h {minutes}m"
 
-    def _format_memory(self, memory_mb: float) -> str:
-        """
-        Format memory usage in human-readable format.
-        Automatically switches between MB and GB based on size.
-
-        Args:
-            memory_mb: Memory usage in megabytes
-
-        Returns:
-            Formatted memory string (e.g., "256 MB" or "1.5 GB")
-        """
-        if memory_mb < 1024:
-            return f"{memory_mb:.1f} MB"
-        else:
-            memory_gb = memory_mb / 1024
-            return f"{memory_gb:.1f} GB"
-
-    def _format_bytes(self, size_bytes: int) -> str:
-        """
-        Format bytes in human-readable format.
-
-        Args:
-            size_bytes: Size in bytes
-
-        Returns:
-            Formatted size string (e.g., "256 bytes", "1.5 kB", "2.0 MB", "1.2 GB")
-        """
-        if size_bytes == 0:
-            return "0 bytes"
-
-        volume = float(size_bytes)
-        unit = "bytes"
-        for unit in ["bytes", "kB", "MB", "GB"]:
-            if volume < 1024 or unit == "GB":
-                break
-            volume /= 1024
-
-        if unit == "bytes":
-            return f"{int(volume)} {unit}"
-        else:
-            return f"{volume:.1f} {unit}"
-
     def check_report(
         self,
         recipe: TamarinRecipe,

src/batch_tamarin/utils/system_resources.py

@@ -138,3 +138,31 @@ def resolve_executable_path(path_or_command: str) -> Path:
         raise FileNotFoundError(f"Command '{path_or_command}' not found in PATH")
 
     return Path(resolved_command)
+
+
+def get_human_readable_volume_size(
+    volume_size: int | float, start_unit: str = "bytes"
+) -> str:
+    """
+    Converts a volume size in bytes into a human-readable unit.
+
+    Args:
+        volume_size (int | float): The volume size in bytes
+        start_unit (str):  One of: 'bytes', 'kB', 'MB', 'GB'
+
+    Returns:
+        str: The human-readable volume size with unit
+    """
+
+    units: list[str] = ["bytes", "kB", "MB", "GB"]
+
+    if start_unit not in units:
+        raise ValueError(f"The value for start_unit must be one of {units}")
+    index = units.index(start_unit)
+
+    for unit in units[index:]:
+        if volume_size < 1024 or unit == "GB":
+            break
+        volume_size /= 1024
+
+    return f"{volume_size:.2f} {unit}"