instrument EPU Agent to send data to Core HTTP API

Author: vredchenko

README.md

@@ -5,7 +5,8 @@
 
 # smartem_decisions
 
-Project board: <https://github.com/orgs/DiamondLightSource/projects/33/views/1>
+- Project board: <https://github.com/orgs/DiamondLightSource/projects/33/views/1>
+- Test Datasets: https://gitlab.diamond.ac.uk/scisoft/cryoem/smartem-decisions-test-datasets
 
 This is where you should write a short paragraph that describes what your module does,
 how it does it, and why people should use it.

docs/software-architecture.mermaid

@@ -44,4 +44,4 @@ graph LR
     class infrastructure infra
     class ext ext
     %% Link styling
-    linkStyle 0,1,2,3,4,5,6,7,8 stroke:#666
+    linkStyle 0,1,2,3,4,5,6,7,8 stroke:#666

setup.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python
 """Setup script to handle version file copying and dotenv instantiation."""
+
 import os
 import shutil
 from pathlib import Path
@@ -16,10 +17,7 @@ def copy_version_files():
         print(f"Warning: {source_file} does not exist, nothing to copy")
         return
 
-    target_files = [
-        Path("src/smartem_decisions/_version.py"),
-        Path("src/epu_data_intake/_version.py")
-    ]
+    target_files = [Path("src/smartem_decisions/_version.py"), Path("src/epu_data_intake/_version.py")]
 
     # Make sure target directories exist
     for target in target_files:
@@ -73,9 +71,9 @@ def run(self):
 setup(
     name="smartem-decisions",  # Required for some setuptools versions
     cmdclass={
-        'develop': CustomDevelop,
-        'install': CustomInstall,
-        'egg_info': CustomEggInfo,
+        "develop": CustomDevelop,
+        "install": CustomInstall,
+        "egg_info": CustomEggInfo,
     },
     # We're using pyproject.toml for the rest
 )

src/epu_data_intake/__main__.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from watchdog.observers import Observer
 
-from src.epu_data_intake.data_model import EpuSession
+from src.epu_data_intake.data_model import EpuAcquisitionSessionStore
 from src.epu_data_intake.fs_parser import EpuParser
 from src.epu_data_intake.fs_watcher import (
     DEFAULT_PATTERNS,
@@ -23,7 +23,7 @@
 @parse_cli.command("dir")
 def parse_epu_output_dir(epu_output_dir: str):
     """Parse an entire EPU output directory structure. May contain multiple grids"""
-    datastore = EpuSession(epu_output_dir)
+    datastore = EpuAcquisitionSessionStore(epu_output_dir)
     datastore = EpuParser.parse_epu_output_dir(datastore)
     logging.info(datastore)
 
@@ -101,42 +101,42 @@ def validate_epu_dir(path: str):
 
 @epu_data_intake_cli.command("watch")
 def watch_directory(
-        path: Path = typer.Argument(..., help="Directory to watch"),
-        patterns: list[str] = typer.Option(
-            DEFAULT_PATTERNS,
-            "--pattern", "-p",
-            help="File patterns to watch (can be specified multiple times)"
-        ),
-        # TODO currently unused because logging should come from `smartem_decisions` module,
-        #  and the log filename should be wired to `log_manager` instantiation once that's in place.
-        log_file: str | None = typer.Option(
-            "fs_changes.log",
-            "--log-file", "-l",
-            help="Log file path (optional)"
-        ),
-        log_interval: float = typer.Option(
-            10.0,
-            "--interval", "-i",
-            help="Minimum interval between log entries in seconds"
-        ),
-        verbose: bool = typer.Option(
-            False,
-            "--verbose", "-v",
-            help="Enable verbose output"
-        ),
+    path: Path = typer.Argument(..., help="Directory to watch"),
+    dry_run: bool = typer.Option(
+        False, "--dry_run", "-n", "Enables dry run mode, writing data in-memory and not posting to Core's HTTP API"
+    ),
+    # TODO provide via env but allow override via CLI, introduce a localhost default for local dev
+    api_url: str = typer.Option(
+        None, "--api-url", help="URL for the Core API (required unless in dry run mode)"
+    ),
+    # TODO currently unused because logging should come from `~smartem_decisions~` `shared` module,
+    #  and the log filename should be wired to `log_manager` instantiation once that's in place.
+    log_file: str | None = typer.Option("fs_changes.log", "--log-file", "-l", help="Log file path (optional)"),
+    log_interval: float = typer.Option(
+        10.0, "--interval", "-i", help="Minimum interval between log entries in seconds"
+    ),
+    patterns: list[str] = typer.Option(
+        DEFAULT_PATTERNS, "--pattern", "-p", help="File patterns to watch (can be specified multiple times)"
+    ),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="Enable verbose output"),
 ):
     """Watch directory for file changes and log them in JSON format."""
     path = Path(path).absolute()
     if not path.exists():
         logging.error(f"Error: Directory {path} does not exist")
         raise typer.Exit(1)
+
+    if not dry_run and not api_url:
+        logging.error("Error: API URL must be provided when not in dry run mode")
+        raise typer.Exit(1)
+
     logging.info(f"Starting to watch directory: {str(path)} (including subdirectories) for patterns: {patterns}")
 
     observer = Observer()
 
     handler = RateLimitedHandler(patterns, log_interval, verbose)
     handler.set_watch_dir(path)
-    handler.init_datastore()
+    handler.init_datastore(dry_run, api_url)
 
     logging.info("Parsing existing directory contents...")
     # TODO settle a potential race condition if one exists:
@@ -155,7 +155,7 @@ def handle_exit(signum, frame):
 
     # Handle both CTRL+C and Windows signals
     signal.signal(signal.SIGINT, handle_exit)
-    if platform.system() == 'Windows':
+    if platform.system() == "Windows":
         signal.signal(signal.SIGBREAK, handle_exit)
 
     observer.start()