wip

Author: vredchenko

src/epu_data_intake/README.md

@@ -98,6 +98,7 @@ python tests/epu_data_intake/epu_output_simulator.py ../test-dir \
 > for instantiating a new grid entity in the internal datastore.
 
 A `watch` operation is designed to gracefully handle one of the following invocation scenarios:
+
 1. watcher launched _before_ EPU starts writing to filesystem - only watcher is necessary
 2. watcher launched _after_ EPU starts writing to filesystem - both parser and watcher are
    necessary to pickup pre-existing and new writes

src/epu_data_intake/__main__.py

@@ -8,11 +8,12 @@
 from pathlib import Path
 from watchdog.observers import Observer
 
+from src.epu_data_intake.core_api_client import SmartEMCoreAPIClient
 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,
-    RateLimitedHandler,
+    RateLimitedFilesystemEventHandler,
 )
 
 # Create a callback to handle the verbose flag at the root level
@@ -174,7 +175,10 @@ def watch_directory(
     dry_run: bool = typer.Option(
         False, "--dry_run", "-n", help="Enables dry run mode, writing data in-memory and not posting to Core's HTTP API"
     ),
-    # TODO consider providing via env but allowing override via CLI. Investigate how Win env vars work or if they do even
+    # TODO
+    #  - consider providing via env but allowing override via CLI.
+    #  - Investigate how Win env vars work.
+    #  - Accept a hard-coded value at build time
     api_url: str = typer.Option(
         "http://127.0.0.1:8000", "--api-url", "-a", help="URL for the Core API (required unless in dry run mode)"
     ),
@@ -196,26 +200,35 @@ def watch_directory(
         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)
+    if not dry_run:
+        try:
+            import asyncio
 
-    logging.info(f"Starting to watch directory: {str(path)} (including subdirectories) for patterns: {patterns}")
+            async def check_api():
+                async with SmartEMCoreAPIClient(api_url) as client:
+                    status_data = await client.get_status()
+                    return status_data
 
-    observer = Observer()
+            status_result = asyncio.run(check_api())
+            logging.info(f"API is reachable at {api_url} - Status: {status_result.get('status', 'unknown')}")
+        except Exception as e:
+            logging.error(f"Error: API at {api_url} is not reachable: {str(e)}")
+            raise typer.Exit(1)
+
+    logging.info(f"Starting to watch directory: {str(path)} (including subdirectories) for patterns: {patterns}")
 
     # TODO verify verbose parameter is still needed If it's *only* used for logging verbosity - check if logging
     #   that's already set up here is being used, making this param redundant
-    handler = RateLimitedHandler(patterns, log_interval, verbose)
-    handler.set_watch_dir(path)
-    handler.init_datastore(dry_run, api_url)
+    handler = RateLimitedFilesystemEventHandler(path, dry_run, api_url, verbose, log_interval, patterns)
 
     logging.info("Parsing existing directory contents...")
-    # TODO settle a potential race condition if one exists:
+    # TODO settle a potential race condition between parser and watcher if one exists:
     handler.datastore = EpuParser.parse_epu_output_dir(handler.datastore, verbose)
     # TODO ^ verify verbose parameter is still needed If it's *only* used for logging verbosity - check if logging
     #   that's already set up here is being used, making this param redundant
+
     logging.info("..done! Now listening for new filesystem events")
+    observer = Observer()
     observer.schedule(handler, str(path), recursive=True)
 
     def handle_exit(signum, frame):

src/epu_data_intake/core_api_client.py

@@ -48,7 +48,7 @@ class MicrographStatus(str, Enum):
 # Request Models
 class AcquisitionCreateRequest(BaseModel):
     name: str
-    epu_id: str | None = None
+    id: str | None = None
     status: AcquisitionStatus | None = None
     start_time: datetime | None = None
     end_time: datetime | None = None
@@ -61,7 +61,7 @@ class AcquisitionCreateRequest(BaseModel):
 
 class AcquisitionUpdateRequest(BaseModel):
     name: str | None = None
-    epu_id: str | None = None
+    id: str | None = None
     status: AcquisitionStatus | None = None
     start_time: datetime | None = None
     end_time: datetime | None = None

src/epu_data_intake/core_api_client_adapter.py

@@ -80,7 +80,7 @@ def create(self, entity_type: str, entity_id: str, entity: T,
                 # Handle acquisition creation - no parent
                 request = AcquisitionCreateRequest(
                     name=entity.name,
-                    epu_id=entity.id,
+                    id=entity.id,
                     start_time=entity.start_time,
                     storage_path=entity.storage_path,
                     atlas_path=entity.atlas_path,

src/epu_data_intake/data_model.py

@@ -4,6 +4,8 @@
 from datetime import datetime
 from pathlib import Path
 from typing import Generic, TypeVar
+from uuid import uuid4
+from base64 import urlsafe_b64encode
 
 T = TypeVar("T")
 
@@ -16,6 +18,10 @@ class EntityStore(Generic[T]):
     _in_memory_only: bool = True
     _parent_entity: tuple[str, str] = None  # (entity_type, entity_id) tuple
 
+    @staticmethod
+    def generate_uuid():
+        return urlsafe_b64encode(uuid4().bytes).decode('ascii').rstrip('=')
+
     def set_api_client(self, api_client, entity_type: str, in_memory_only: bool = True):
         self._api_client = api_client
         self._entity_type = entity_type
@@ -24,24 +30,25 @@ def set_api_client(self, api_client, entity_type: str, in_memory_only: bool = Tr
     def set_parent_entity(self, parent_type: str, parent_id: str):
         self._parent_entity = (parent_type, parent_id)
 
-    def add(self, id: str, item: T) -> None:
-        is_update = id in self._items
-        self._items[id] = item
+    def add(self, uuid: str | None, item: T) -> None:
+        if not uuid: uuid = EntityStore.generate_uuid()
+        is_new = uuid not in self._items
+        self._items[uuid] = item
 
         if not self._in_memory_only and self._api_client:
             try:
-                if is_update:
-                    self._api_client.update(self._entity_type, id, item, self._parent_entity)
+                if is_new:
+                    self._api_client.create(self._entity_type, uuid, item, self._parent_entity)
                 else:
-                    self._api_client.create(self._entity_type, id, item, self._parent_entity)
+                    self._api_client.update(self._entity_type, uuid, item, self._parent_entity)
             except Exception as e:
-                logging.error(f"API sync failed for {self._entity_type}/{id}: {str(e)}")
+                logging.error(f"API sync failed for {self._entity_type}/{uuid}: {str(e)}")
 
-    def get(self, id: str) -> T | None:
-        return self._items.get(id)
+    def get(self, uuid: str) -> T | None:
+        return self._items.get(uuid)
 
-    def exists(self, id: str) -> bool:
-        return id in self._items
+    def exists(self, uuid: str) -> bool:
+        return uuid in self._items
 
     def ids(self) -> set[str]:
         return set(self._items.keys())
@@ -256,9 +263,9 @@ class AtlasData:
 
 @dataclass
 class EpuSessionData:
-    name: str
-    id: str
-    start_time: datetime
+    name: str = "Unknown"
+    id: str = EntityStore.generate_uuid()
+    start_time: datetime | None = None
     atlas_path: str | None = None
     storage_path: str | None = None  # Path of parent directory containing the epu session dir
     clustering_mode: str | None = None
@@ -307,19 +314,22 @@ class EpuAcquisitionSessionStore:
     root_dir: Path
     in_memory_only: bool = False
     api_client = None
+    uuid: str | None = None
 
-    def __init__(self, root_dir: str, dry_run: bool = False, api_url: str = None):
+    def __init__(self, root_dir: str, in_memory_only: bool = False, api_url: str = None):
         self.root_dir = Path(root_dir)
-        self.in_memory_only = dry_run
+        self.in_memory_only = in_memory_only
+        self.acquisition = EpuSessionData()
         self.grids = EntityStore()
 
-        if not dry_run and api_url:
+        if not self.in_memory_only:
             from src.epu_data_intake.core_api_client_adapter import ApiClientAdapter
             self.api_client = ApiClientAdapter(api_url)
-            self.grids.set_api_client(self.api_client, "grid", dry_run)
+            self.api_client.create("acquisition", self.acquisition.id, self.acquisition, None)
+            self.grids.set_api_client(self.api_client, "grid", in_memory_only)
 
-    def add_grid(self, grid_id: str, grid: Grid):
-        """Add a grid with proper API configuration"""
+    def add_grid(self, grid: Grid, grid_id: str | None = None):
+        grid_id = grid_id or EntityStore.generate_uuid()
 
         # Configure the grid's EntityStores for the hierarchy:
         # session -> grid -> gridsquare -> foilhole -> micrograph
@@ -363,7 +373,7 @@ def get_grid_by_path(self, path: str):
         return None
 
     def __str__(self):
-        result = f"\nEPU Acquisition Summary:\n  Root dir: {self.root_dir}\n  Grids: {len(self.grids)}\n"
+        result = f"\nEPU Acquisition Summary:\n  UUID: {self.uuid}\n Root dir: {self.root_dir}\n  Grids: {len(self.grids)}\n"
 
         # Add each grid's string representation. TODO debug this doesn't seem to work
         for grid_id, grid in self.grids.items():

src/epu_data_intake/fs_parser.py

@@ -695,7 +695,7 @@ def parse_epu_output_dir(datastore: EpuAcquisitionSessionStore, verbose: bool =
         # Start with locating all EpuSession.dm files - init a grid for each found
         for epu_session_manifest in list(datastore.root_dir.glob("**/*EpuSession.dm")):
             grid = EpuParser.parse_grid_dir(str(Path(epu_session_manifest).parent), verbose)
-            datastore.grids.add(grid.session_data.name, grid)
+            datastore.add_grid(grid)
 
         return datastore
 

src/epu_data_intake/fs_watcher.py

@@ -41,7 +41,7 @@
 ]
 
 
-class RateLimitedHandler(FileSystemEventHandler):
+class RateLimitedFilesystemEventHandler(FileSystemEventHandler):
     """File system event handler with rate limiting capabilities.
 
     This handler processes file system events from watchdog, specifically watching
@@ -70,7 +70,8 @@ class RateLimitedHandler(FileSystemEventHandler):
     datastore: EpuAcquisitionSessionStore | None = None
     watch_dir: Path | None = None
 
-    def __init__(self, patterns: list[str], log_interval: float = 10.0, verbose: bool = False):
+    # TODO test with a lower log_interval value, set lowest possible default, better naming
+    def __init__(self, watch_dir, dry_run: bool = False, api_url: str | None = None, verbose: bool = False, log_interval: float = 10.0, patterns: list[str] = DEFAULT_PATTERNS):
         self.last_log_time = time.time()
         self.log_interval = log_interval
         self.patterns = patterns
@@ -82,7 +83,10 @@ def __init__(self, patterns: list[str], log_interval: float = 10.0, verbose: boo
         # Maintain a buffer of "orphaned" files - files that appear to belong to a grid that doesn't exist yet
         self.orphaned_files = {}  # path -> (event, timestamp, file_stat)
 
-    # TODO unit test thi method
+        self._set_watch_dir(watch_dir)
+        self._init_datastore(dry_run, api_url)
+
+    # TODO unit test this method
     def matches_pattern(self, path: str) -> bool:
         try:
             rel_path = str(Path(path).relative_to(self.watch_dir))
@@ -91,29 +95,28 @@ def matches_pattern(self, path: str) -> bool:
         except ValueError:
             return False
 
-    def set_watch_dir(self, path: Path):
-        self.watch_dir = path.absolute()  # TODO this could cause problems in Win
+    # TODO on Win there's primary and secondary output dirs - work directly with primary if possible otherwise
+    #  operate across both. Note: data is first written to primary output dir then later maybe partially copied
+    #  to secondary dir.
+    def _set_watch_dir(self, path: Path):
+        self.watch_dir = path.absolute()  # TODO this could cause problems in Win - test!
 
-    def init_datastore(self, dry_run: bool = False, api_url: str = None):
-        logging.info(f"Instantiated new datastore, dry run: {dry_run}")
+    def _init_datastore(self, dry_run: bool = False, api_url: str = None):
         self.datastore = EpuAcquisitionSessionStore(str(self.watch_dir), dry_run, api_url)
+        logging.debug(f"Instantiated new datastore, " + ("in-memory only" if self.datastore.in_memory_only else "data will be permanently saved on the backend"))
+        if self.datastore.in_memory_only:
+            logging.info(f"Acquisition session uuid assigned: {self.datastore.uuid}")
 
+    # TODO Enhancement: log all events to graylog (if reachable) for session debugging and playback
     def on_any_event(self, event):
-        if self.watch_dir is None:
-            raise RuntimeError("watch_dir not initialized - call set_watch_dir() first")
-        if self.datastore is None:
-            raise RuntimeError("datastore not initialized - call init_datastore() first")
-
-        # Enhancement: record all events to graylog (if reachable) for session debugging and playback
-
         if event.is_directory or not self.matches_pattern(event.src_path):
             if event.is_directory:
-                logging.info(f"Skipping non-matching path: {event.src_path}")
+                logging.debug(f"Skipping non-matching path: {event.src_path}")
             return
 
         if event.event_type not in self.watched_event_types:
             if event.is_directory:
-                logging.info(f"Skipping non-matching event type: {event.event_type}")
+                logging.debug(f"Skipping non-matching event type: {event.event_type}")
             return
 
         current_time = time.time()
@@ -135,18 +138,15 @@ def on_any_event(self, event):
         if new_file_detected and re.search(EpuParser.session_dm_pattern, event.src_path):
             assert self.datastore.get_grid_by_path(event.src_path) is None  # guaranteed because is a new file
             grid = Grid(str(Path(event.src_path).parent.resolve()))
-            session_data = EpuParser.parse_epu_session_manifest(
-                str(Path(event.src_path).resolve())
-            )  # just to get the name really, techdebt
-            self.datastore.grids.add(session_data.name, grid)
+            self.datastore.add_grid(grid)
 
         # try to work out which grid the touched file relates to
         grid_id = self.datastore.get_grid_by_path(event.src_path)
         if grid_id is None:
             # This must be an orphaned file since it matched one of patterns for files we are interested in,
             #   but a containing grid doesn't exist yet - store it for when we have the grid.
             if self.verbose:
-                logging.info(
+                logging.debug(
                     f"Could not determine which grid this data belongs to: {event.src_path}, adding to orphans"
                 )
             self.orphaned_files[event.src_path] = (event, current_time, file_stat)
@@ -169,27 +169,27 @@ def on_any_event(self, event):
                 self._on_micrograph_detected(path, grid_id, new_file_detected)
 
     def _on_session_detected(self, path: str, grid_id, is_new_file: bool = True):
-        logging.info(f"Session manifest {'detected' if is_new_file else 'updated'}: {path}")
+        logging.debug(f"Session manifest {'detected' if is_new_file else 'updated'}: {path}")
         session_data = EpuParser.parse_epu_session_manifest(path)
         gridstore = self.datastore.grids.get(grid_id)
 
         if gridstore and session_data != gridstore.session_data:
             # Create the acquisition first if we're not in dry run mode
             if not self.datastore.in_memory_only and self.datastore.api_client:
-                success = self.datastore.api_client.create("acquisition", session_data.id, session_data)
+                success = self.datastore.api_client.create("acquisition", self.datastore.uuid, session_data)
                 if success:
-                    logging.info(f"Created acquisition for session {session_data.name}")
+                    logging.info(f"Created acquisition for session #{self.datastore.uuid} {session_data.name}")
 
             gridstore.session_data = session_data
-            logging.info(f"Updated session data for grid {grid_id}")
+            logging.debug(f"Updated session data for grid {grid_id}")
             logging.info(gridstore.session_data)
 
     def _process_orphaned_files(self, grid_id):
         """Process any orphaned files that belong to this grid"""
         for path, (event, timestamp, file_stat) in self.orphaned_files.items():
             # Check if this orphaned file belongs to the new grid
             if self.datastore.get_grid_by_path(path) == grid_id:
-                logging.info(f"Processing previously orphaned file: {path}")
+                logging.debug(f"Processing previously orphaned file: {path}")
                 self.on_any_event(event)  # Process the file as if we just received the event
 
         # Create a new dictionary excluding the processed files
@@ -198,15 +198,15 @@ def _process_orphaned_files(self, grid_id):
         }
 
     def _on_atlas_detected(self, path: str, grid_id, is_new_file: bool = True):
-        logging.info(f"Atlas {'detected' if is_new_file else 'updated'}: {path}")
+        logging.debug(f"Atlas {'detected' if is_new_file else 'updated'}: {path}")
         gridstore = self.datastore.grids.get(grid_id)
         atlas_data = EpuParser.parse_atlas_manifest(path)
         if atlas_data != gridstore.atlas_data:
             gridstore.atlas_data = atlas_data
             logging.info(gridstore.atlas_data)
 
     def _on_gridsquare_metadata_detected(self, path: str, grid_id, is_new_file: bool = True):
-        logging.info(f"Gridsquare metadata {'detected' if is_new_file else 'updated'}: {path}")
+        logging.debug(f"Gridsquare metadata {'detected' if is_new_file else 'updated'}: {path}")
 
         gridsquare_id = EpuParser.gridsquare_dm_file_pattern.search(path).group(1)
         assert gridsquare_id is not None, f"gridsquare_id should not be None: {gridsquare_id}"