Merge pull request #10 from i-VRESSE/time-checks

Add recorded_argparse decorator

Author: sverhoeven

AGENTS.md

@@ -7,7 +7,7 @@ uvx ruff format          # Format code
 uvx ruff check           # Check for issues
 uv run pyright           # Type check code
 uv build                 # Build package
-uv run --group docs sphinx-build docs docs/_build  # Build documentation
+uv run --group docs sphinx-build docs docs/_build  # Build documentation in docs/_build
 ```
 
 ## Style

README.md

@@ -19,44 +19,39 @@ pip install rocrate-action-recorder
 
 ## Usage
 
-Shown is an example of recording a CLI command (`example-cli input.txt output.txt`) implemented with `argparse`.
+Example of recording a CLI command (`example-cli input.txt output.txt`) implemented with `argparse`.
 
 ```python
 import argparse
-from datetime import datetime
 from pathlib import Path
-from rocrate_action_recorder import record_with_argparse, IOArgumentNames
+import sys
+
+from rocrate_action_recorder import recorded_argparse
 
-# Create an argparse parser
 parser = argparse.ArgumentParser(prog="example-cli", description="Example CLI")
-parser.add_argument("--version", action="version", version="1.2.3")
+parser.add_argument("--version", action="version", version="%(prog)s 1.2.3")
+parser.add_argument("--no-record", action="store_false", help="Disable RO-Crate recording.")
 parser.add_argument("input", type=Path, help="Input file")
 parser.add_argument("output", type=Path, help="Output file")
 
+@recorded_argparse(
+    parser=parser,
+    input_files=["input"],
+    output_files=["output"],
+    dataset_license="CC-BY-4.0",
+    enabled_argument="no_record",
+)
+def handler(args: argparse.Namespace):
+    # For demonstration, just upper case input to output, replace with real logic
+    args.output.write_text(args.input.read_text().upper())
+
 # Prepare input
 Path("input.txt").write_text("hello")
+# Simulate command-line arguments
+sys.argv.extend(["input.txt", "output.txt"])
 
-# Parse arguments
-args = ['input.txt', 'output.txt']
-ns = parser.parse_args(args)
-
-# Do handling of the CLI command here
-start_time = datetime.now()
-# For demonstration, just upper case input to output
-Path(ns.output).write_text(ns.input.read_text().upper())
-
-record_with_argparse(
-    parser, 
-    ns, 
-    # Tell recorder which arguments are for input and output files
-    IOArgumentNames(input_files=["input"], output_files=["output"]),
-    start_time, 
-    dataset_license="CC-BY-4.0",
-    # argv argument is optional, in real usage you can omit it
-    argv=['example-cli'] + args,
-    # current_user argument is optional, in real usage you can omit it
-    current_user="someuser"
-)
+args = parser.parse_args()
+handler(args)
 ```
 
 <details>
@@ -161,9 +156,9 @@ Will generate a `ro-crate-metadata.json` file in the current working directory d
 }
 ```
 
-</details>
-
+For you the startTime, endTime, and Person name will differ.
 
+</details>
 
 <details>
 <summary>

docs/index.md

@@ -11,8 +11,8 @@
 :end-before: <!-- SPHINX-END -->
 ```
 
-The main functions are:
-- [record_from_argparse](#rocrate_action_recorder.adapters.argparse.record_with_argparse) function.
+The main entry points are:
+- [recorded_argparse](#rocrate_action_recorder.adapters.argparse.recorded_argparse) decorator.
 - [record](#rocrate_action_recorder.core.record) function.
 
 ## Example

example/myscript.py

@@ -1,10 +1,11 @@
 #!/usr/bin/env python3
 import argparse
-from datetime import datetime
+from functools import lru_cache
 from pathlib import Path
-from rocrate_action_recorder import record_with_argparse, IOs
+from rocrate_action_recorder import recorded_argparse
 
 
+@lru_cache(maxsize=1)  # Cache the parser instance to avoid re-creating it for each handler
 def make_parser():
     parser = argparse.ArgumentParser(prog="myscript", description="Example CLI")
     parser.add_argument("--version", action="version", version="%(prog)s 1.0.0")
@@ -13,25 +14,21 @@ def make_parser():
     return parser
 
 
-def handler(args, parser):
-    start_time = datetime.now()
+@recorded_argparse(
+    parser=make_parser(),
+    input_files=["input"],
+    output_files=["output"],
+    dataset_license="CC-BY-4.0",
+)
+def handler(args: argparse.Namespace) -> int:
     # do something simple
-    args.output.write_text(args.input.read_text().upper())
-
-    ios = IOs(input_files=["input"], output_files=["output"])
-    record_with_argparse(
-        parser=parser,
-        ns=args,
-        ios=ios,
-        start_time=start_time,
-        dataset_license="CC-BY-4.0",
-    )
+    return args.output.write_text(args.input.read_text().upper())
 
 
 def main():
     parser = make_parser()
     args = parser.parse_args()
-    handler(args, parser)
+    handler(args)
 
 
 if __name__ == "__main__":

src/rocrate_action_recorder/__about__.py

@@ -1 +1 @@
-__version__ = "0.1.0"
+__version__ = "0.2.0"

src/rocrate_action_recorder/__init__.py

@@ -1,6 +1,10 @@
 """RO-Crate action recorder for CLI invocations."""
 
-from rocrate_action_recorder.adapters.argparse import record_with_argparse, IOArgumentNames
+from rocrate_action_recorder.adapters.argparse import (
+    record_argparse,
+    recorded_argparse,
+    IOArgumentNames,
+)
 from rocrate_action_recorder.core import (
     IOArgumentPath,
     IOArgumentPaths,
@@ -10,7 +14,8 @@
 )
 
 __all__ = [
-    "record_with_argparse",
+    "record_argparse",
+    "recorded_argparse",
     "record",
     "playback",
     "Program",

src/rocrate_action_recorder/adapters/argparse.py

@@ -1,8 +1,10 @@
 """Adapter for argparse CLI framework."""
 
 from argparse import _VersionAction, ArgumentParser, Namespace
+from collections.abc import Callable
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
+from functools import wraps
 from pathlib import Path
 from typing import Any
 import logging
@@ -235,7 +237,7 @@ def collect_record_info_from_argparse(
     return program, ioargs
 
 
-def record_with_argparse(
+def record_argparse(
     parser: ArgumentParser,
     ns: Namespace,
     ios: IOArgumentNames,
@@ -288,3 +290,70 @@ def record_with_argparse(
         current_user=current_user,
         dataset_license=dataset_license,
     )
+
+
+def recorded_argparse[T](
+    parser: ArgumentParser,
+    input_dirs: list[str] | None = None,
+    output_dirs: list[str] | None = None,
+    input_files: list[str] | None = None,
+    output_files: list[str] | None = None,
+    dataset_license: str | None = None,
+    enabled_argument: str | None = None,
+) -> Callable[[Callable[[Namespace], T]], Callable[[Namespace], T]]:
+    """Decorator to record a CLI invocation in an RO-Crate using argparse.
+
+    Args:
+        parser: The argument parser used to parse the command-line arguments.
+            This is needed to extract program information and help texts for the arguments.
+        input_dirs: List of argument names representing input directories
+        output_dirs: List of argument names representing output directories
+        input_files: List of argument names representing input files
+        output_files: List of argument names representing output files
+        dataset_license: License string for the dataset (e.g., "CC BY 4.0").
+                        If None, no license is recorded.
+        enabled_argument: Name of the attribute in args that indicates whether
+                       to record the invocation. Records if None.
+                       If provided, the invocation is only recorded if getattr(args, enabled_argument) is truthy.
+
+    Returns:
+        Decorator function
+
+    Raises:
+        ValueError:
+            If the current user cannot be determined.
+            If the specified paths are outside the crate root.
+            If the software version cannot be determined based on the program name.
+        MissingDestArgparseSubparserError:
+            If parser has subparsers but dest is not set.
+    """
+
+    def decorator(func: Callable[[Namespace], T]) -> Callable[[Namespace], T]:
+        @wraps(func)
+        def wrapper(args: Namespace) -> T:
+            start_datetime = datetime.now(tz=UTC)
+
+            result = func(args)
+
+            if enabled_argument is None or getattr(args, enabled_argument, False):
+                end_time = datetime.now(tz=UTC)
+                ios = IOArgumentNames(
+                    input_dirs=input_dirs or [],
+                    output_dirs=output_dirs or [],
+                    input_files=input_files or [],
+                    output_files=output_files or [],
+                )
+                record_argparse(
+                    parser=parser,
+                    ns=args,
+                    ios=ios,
+                    start_time=start_datetime,
+                    end_time=end_time,
+                    dataset_license=dataset_license,
+                )
+
+            return result
+
+        return wrapper
+
+    return decorator

src/rocrate_action_recorder/core.py

@@ -1,7 +1,7 @@
 """Core functionality for recording CLI invocations in RO-Crate format."""
 
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
 import getpass
 import importlib.metadata
 import inspect
@@ -200,6 +200,8 @@ def record(
             If the current user cannot be determined.
             If the specified paths are outside the crate root.
             If the software version cannot be determined based on the program name.
+            If start_time and end_time have different timezone information.
+            If start_time is after end_time.
     """
     crate_root = Path(crate_dir or Path.cwd()).resolve().expanduser()
     crate_root.mkdir(parents=True, exist_ok=True)
@@ -213,7 +215,16 @@ def record(
             current_user = getpass.getuser()
 
     if end_time is None:
-        end_time = datetime.now()
+        end_time = datetime.now(tz=UTC)
+
+    if start_time.tzinfo != end_time.tzinfo:
+        raise ValueError(
+            f"start_time and end_time must have same timezone. Now start_time tzinfo is {start_time.tzinfo} and end_time tzinfo is {end_time.tzinfo}"
+        )
+    if start_time > end_time:
+        raise ValueError(
+            f"start_time must be before end_time. Now start_time is {start_time} and end_time is {end_time}"
+        )
 
     if not program.version:
         program.version = detect_software_version(program.name)