From 783597f8eef2f18095bc73b9d34c1775e30e944f Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Wed, 8 Oct 2025 16:04:54 -0500 Subject: [PATCH 01/11] Adding listener files --- configs/watchdog_listener_config.toml | 63 ++++++++ listen.py | 44 ++++++ source/listeners/base.py | 216 ++++++++++++++++++++++++++ source/listeners/watchdog_listener.py | 125 +++++++++++++++ source/uploaders/local_to_s3.py | 2 +- 5 files changed, 449 insertions(+), 1 deletion(-) create mode 100644 configs/watchdog_listener_config.toml create mode 100644 listen.py create mode 100644 source/listeners/base.py create mode 100644 source/listeners/watchdog_listener.py diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml new file mode 100644 index 00000000..92e27b77 --- /dev/null +++ b/configs/watchdog_listener_config.toml @@ -0,0 +1,63 @@ +# ======================================================= +# Config for the Watchdog Listener +# ======================================================= + +[parser] + +[parser.class] +type = "dynamic" +name = "LocalToS3BatchListener" +parts = [ + ["listeners.watchdog_listener", "WatchdogListenerMixin"], + ["uploaders.local_to_s3", "S3UploaderMixin"] +] + + +[parser.init] +# The directory where the state file ('upload_state.json') will be stored. +state_path = "C:/Users/Nihil/datanet_files/upload_state.json" + + +[parser.init.source] +# The local directory that the listener will monitor for new files. +path = "E:/local" +#File name patterns to include/exclude. +include_patterns = [".*\\.csv$", ".*\\.txt$"] +exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] +#Batch configuration +# Trigger the pipeline after 5 files have accumulated... +batch_max_size = 5 +# ...or after 60 seconds have passed since the first file arrived, +# whichever comes first. +batch_max_latency_seconds = 60 + +[parser.init.target] +# Settings for the S3 uploader. +bucket = "data-net-source-debug" +prefix = "raw_data/" +# Set to 'true' to upload files that have been modified. +allow-overwrite = true + + + +# Enable logging to a local file using the python's built-in logging. +[logging.file] +filepath = "C:/Users/Nihil/datanet_files/s3_parser_log.txt" +level = 0 # level of events to include in this logger +max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out +max_files = 5 # Max number of files to keep before deleting the oldest + +# Enable logging to Sentry using the Sentry SDK API +# [logging.sentry] +# dsn = "placeholder" +# level = 10 +# event_level = 30 + +[dependencies] +# Required dependencies for this specific parser configuration. +# The `prepare.py` script will install these. +pip = [ + "boto3", + "watchdog" +] + diff --git a/listen.py b/listen.py new file mode 100644 index 00000000..49a33612 --- /dev/null +++ b/listen.py @@ -0,0 +1,44 @@ +""" +Entry-point script to load and run a parser in continuous, real-time listener mode. +""" + +import argparse + +# Reuse the existing functions from run.py to load the config and build the parser. +from run import load_config, load_parser + + +if __name__ == '__main__': + # 1. Set up argument parsing to accept the path to a config file. + arg_parser = argparse.ArgumentParser( + description="Run a data-net-source parser in continuous listener mode." + ) + arg_parser.add_argument( + 'config_file', + type=str, + help='Path to the .toml config file that specifies the parser to run.' + ) + args = arg_parser.parse_args() + + # 2. Load the specified configuration file using the helper from run.py. + config = load_config(args.config_file) + + # 3. Dynamically build the parser object from the mixins defined in the config. + parser = load_parser(config['parser']) + + # 4. Set up logging as defined in the config, if present. + if 'logging' in config: + parser.make_loggers(config['logging']) + + # 5. Call the .listen() method to start the long-running, event-driven process. + try: + parser.listen() + except AttributeError: + # Provide a helpful error if the user tries to run a checker-based config + # with this script, as it will not have a .listen() method. + parser.error( + f"The configured parser '{parser.__class__.__name__}' does not have a .listen() method. " + f"Ensure your config file specifies a Listener mixin, not a Checker mixin." + ) + except Exception as e: + parser.error(f"The listener exited with an unexpected error: {e}", exc_info=True) \ No newline at end of file diff --git a/source/listeners/base.py b/source/listeners/base.py new file mode 100644 index 00000000..51100813 --- /dev/null +++ b/source/listeners/base.py @@ -0,0 +1,216 @@ +import time +import threading +import copy +import os +import json +from datetime import datetime +from abc import ABC, abstractmethod + +from source.common import EMPTY_LOG + + +class BaseListener(ABC): + """ + Abstract base class for all listener mixins. + + This class defines the required interface for a listener component, thread-safe batching and provides + the common logic for processing events and managing state. + """ + state_filename = 'upload_state.json' + state_path = "path/to/state/save/dir" + + # ------------------------------------------------------------------------- + # --- Abstract methods to be implemented by concrete listeners --- + # ------------------------------------------------------------------------- + + @property + @abstractmethod + def listener_name(self) -> str: + """A simple attribute naming the mixin class for later reference.""" + pass + + @abstractmethod + def listen(self): + """ + Start the listener. + + This should be a blocking call that runs continuously to monitor an + event source (e.g., start a watchdog observer, connect to a message queue). + """ + pass + + @abstractmethod + def _parse_event(self, raw_event: any) -> dict: + """ + Parse a raw event from the source into the standard task format. + + This is the primary abstraction point. It translates a source-specific + event object into the dictionary format expected by the `transform` method. + + :param raw_event: The native event object from the source (e.g., a + watchdog event, a message from a queue). + :return: A dictionary in the format {'to do': [...], 'failure': [...]}. + Return an empty or None dict to indicate the event should be skipped. + """ + pass + + @abstractmethod + def save(self, completed: dict): + """ + Save the state of processed files. + + :param completed: A dict with 'success', 'failure', and 'skipped' lists. + """ + pass + + @abstractmethod + def clean(self): + """Perform cleanup actions, like removing old files or consolidating state entries.""" + pass + + # ------------------------------------------------------------------------- + # --- Concrete methods providing the batching functionality --- + # ------------------------------------------------------------------------- + + def _setup_batching(self): + """Initializes all common batching attributes from the configuration.""" + source_config = self.source_location + self._batch_max_size = source_config.get('batch_max_size', 10) + self._batch_max_latency_seconds = source_config.get('batch_max_latency_seconds', 60) + self._file_buffer = [] + self._in_flight = set() + self._buffer_lock = threading.Lock() + self._batch_timer = None + self.info(f"Batching configured with max size: {self._batch_max_size} and max latency: {self._batch_max_latency_seconds}s") + + def add_to_batch(self, raw_event: any): + """ + Adds a file from an event to the batch buffer, handles renames, + and checks batch triggers using a unified logic. + """ + final_path, old_path = self._parse_event(raw_event) + if not final_path: + return # The event was ignored by the parser. + + # Before locking, check if this file is already being processed. + if final_path in self._in_flight: + self.debug(f"Ignoring event for in-flight file: {final_path}") + return + + process_now = False + with self._buffer_lock: + # If the event was a move, try to remove the old path from the buffer. + if old_path and old_path in self._file_buffer: + self._file_buffer.remove(old_path) + + # Add the new/final path, but only if it's not already present. + if final_path not in self._file_buffer: + self._file_buffer.append(final_path) + self.info(f"Added to buffer: {final_path}. Current size: {len(self._file_buffer)}") + + if not self._file_buffer: + return + + # Start the timer only if this is the first item in a new batch. + if len(self._file_buffer) == 1: + self._batch_timer = threading.Timer( + self._batch_max_latency_seconds, self._process_batch + ) + self._batch_timer.start() + + # If the buffer is full, set a flag to process the batch. + if len(self._file_buffer) >= self._batch_max_size: + process_now = True + + if process_now: + self._process_batch() + + def _process_batch(self): + """Processes all files currently in the buffer.""" + files_to_process = [] + with self._buffer_lock: + if not self._file_buffer: + return + + if self._batch_timer and self._batch_timer.is_alive(): + self._batch_timer.cancel() + + files_to_process = self._file_buffer.copy() + self._file_buffer.clear() + # Add the files to the in-flight set to avoid duplicates. + self._in_flight.update(files_to_process) + + self.info(f"Processing batch of {len(files_to_process)} files...") + try: + tasks = {'to do': files_to_process, 'failure': []} + self.info(f"Batch of {len(files_to_process)} files processed successfully to the next stage.") + ready = self.transform(tasks) + completed = self.upload(ready) + self.save(completed) + self.clean() + except Exception as e: + self.error(f"Failed to process batch: {e}", exc_info=True) + finally: + # Always remove the files from the in-flight set after processing. + self._in_flight.difference_update(files_to_process) + + # ------------------------------------------------------------------------- + # --- State Management Helper Methods (from BaseChecker) --- + # ------------------------------------------------------------------------- + + def load_state(self) -> dict: + """Load the saved upload state from a previously saved file.""" + with open(os.path.join(self.state_path, self.state_filename)) as f: + log = json.load(f) + return log + + def write_state(self, state_data: dict): + """Write the passed state of successes and failures to file.""" + with open(os.path.join(self.state_path, self.state_filename), 'w') as log: + json.dump(state_data, log, indent=2) + + def clean_outdated(self, full_state: dict) -> dict: + """Remove all entries for a particular source file except for the most recent one.""" + def iter_state(key, state_dict): + for obj in state_dict.get(key, []): + if 'filename' in obj: + yield obj['filename'], obj + elif 'uploaded' in obj: + yield obj['uploaded'], obj + + re_organized = {} + for category in full_state: + for filename, event in iter_state(category, full_state): + if filename not in re_organized: + re_organized[filename] = [] + re_organized[filename].append((category, event)) + + reduced = copy.deepcopy(EMPTY_LOG) + for filename, entries in re_organized.items(): + if len(entries) > 1: + entries = sorted(entries, key=lambda x: x[1]['timestamp']) + cat, event = entries[-1] + reduced[cat].append(event) + + return reduced + + def clean_old_success(self, state: dict) -> dict: + """Delete local files that have been successfully uploaded long enough ago.""" + successes = state.get('success', []) + kept_success = [] + now = datetime.now().timestamp() + has_delete = hasattr(self, 'delete_age_hours') + for uploaded in successes: + age = (now - uploaded['timestamp']) / (60 * 60) + if has_delete and self.delete_age_hours >= 0 and age > self.delete_age_hours: + self.info(f'Deleting {uploaded.get("uploaded")}') + try: + os.remove(uploaded.get("uploaded")) + except (FileNotFoundError, TypeError): + self.warning(f'File was already deleted or path was invalid!') + else: + kept_success.append(uploaded) + + new_log = copy.deepcopy(state) + new_log['success'] = kept_success + return new_log \ No newline at end of file diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py new file mode 100644 index 00000000..4601b508 --- /dev/null +++ b/source/listeners/watchdog_listener.py @@ -0,0 +1,125 @@ +import os +import time +import threading +from datetime import datetime +from watchdog.observers import Observer +from watchdog.events import ( + FileSystemEventHandler, + FileClosedEvent, + FileMovedEvent, + FileCreatedEvent, + FileModifiedEvent, + RegexMatchingEventHandler +) + +from source.listeners.base import BaseListener + + +class PipelineEventHandler(RegexMatchingEventHandler): + """ + A custom event handler that bridges watchdog and the data-net-source pipeline. + """ + def __init__(self, parser_instance, **kwargs): + # Pass all regex arguments to the superclass for filtering. + super().__init__(**kwargs) + self.parser = parser_instance + self.parser.info("RegexMatchingEventHandler initialized. Ready for matched events.") + + def on_created(self, event: FileCreatedEvent): + """Called when a file or directory is created.""" + self.parser.add_to_batch(event) + + def on_modified(self, event: FileModifiedEvent): + """Called when a file is modified.""" + self.parser.add_to_batch(event) + + def on_closed(self, event: FileClosedEvent): + """Called when a file opened for writing is closed.""" + self.parser.add_to_batch(event) + + def on_moved(self, event: FileMovedEvent): + """Called when a file is moved or renamed.""" + self.parser.add_to_batch(event) + + +class WatchdogListenerMixin(BaseListener): + """ + A concrete listener that uses 'watchdog' to monitor a directory and processes + new files. + """ + @property + def listener_name(self) -> str: + return "WatchdogBatchListener" + + def listen(self): + """Initializes batching and starts the watchdog observer with the regex handler.""" + self._setup_batching() + source_config = self.source_location + path = self.source_location.get('path') + if not path: + raise ValueError("[parser.init.source] must contain a 'path' key.") + + # Preparing arguments for the RegexMatchingEventHandler from the config file. + handler_kwargs = { + 'regexes': source_config.get('include_patterns', ['.*']), # Default to match everything + 'ignore_regexes': source_config.get('exclude_patterns', []), + 'ignore_directories': True, # Using the handler's built-in directory ignoring - helps empty directories + 'case_sensitive': False + } + + self.info(f"Starting file system listener on directory: {path}") + event_handler = PipelineEventHandler(parser_instance=self, **handler_kwargs) + observer = Observer() + observer.schedule(event_handler, path, recursive=True) + observer.start() + self.start_notify() + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + self.warning("Listener stopped by user. Processing any remaining files...") + self._process_batch() + self.end_notify(0) + finally: + observer.stop() + observer.join() + + def _parse_event(self, raw_event: FileSystemEventHandler) -> tuple[str | None, str | None]: + """ + Parses a watchdog event to extract file paths and filter unwanted files, + returning a (final_path, old_path) tuple. Path-based filtering is + handled by the RegexMatchingEventHandler. + """ + # Ignore all events that are for directories, double checking here. + if raw_event.is_directory: + return None, None + + final_path, old_path = None, None + if isinstance(raw_event, FileMovedEvent): + final_path = raw_event.dest_path + old_path = raw_event.src_path + else: + final_path = raw_event.src_path + + return final_path, old_path + + def save(self, completed: dict): + """Saves the state for a completed batch of files.""" + now = datetime.now().timestamp() + for status, events in completed.items(): + for event in events: + event['timestamp'] = now + + current_state = self.load_state() + for key in ['success', 'failure', 'skipped']: + if key in completed and completed[key]: + current_state[key].extend(completed[key]) + self.write_state(current_state) + + def clean(self): + """Performs state file cleanup after a batch is processed.""" + current_state = self.load_state() + deduplicated_state = self.clean_outdated(current_state) + final_state = self.clean_old_success(deduplicated_state) + self.write_state(final_state) \ No newline at end of file diff --git a/source/uploaders/local_to_s3.py b/source/uploaders/local_to_s3.py index 13a71be8..aa4d74e4 100644 --- a/source/uploaders/local_to_s3.py +++ b/source/uploaders/local_to_s3.py @@ -1,4 +1,4 @@ -# source/uploaders/s3.py +# source/uploaders/local_to_s3.py import boto3 import os From 7ee2b95515cdad269ea25528d1f4250b0f696f8c Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Fri, 10 Oct 2025 02:21:46 -0500 Subject: [PATCH 02/11] Abstracted oberver handler to base listener --- source/listeners/base.py | 54 +++++++++++++++++++++++++++ source/listeners/watchdog_listener.py | 35 ++++++----------- 2 files changed, 65 insertions(+), 24 deletions(-) diff --git a/source/listeners/base.py b/source/listeners/base.py index 51100813..880fb58b 100644 --- a/source/listeners/base.py +++ b/source/listeners/base.py @@ -6,6 +6,7 @@ from datetime import datetime from abc import ABC, abstractmethod +from watchdog.observers import Observer from source.common import EMPTY_LOG @@ -68,6 +69,59 @@ def clean(self): """Perform cleanup actions, like removing old files or consolidating state entries.""" pass + @abstractmethod + def _create_event_handler(self): + """ + Create the specific event handler for this listener type. + + :return: Event handler instance compatible with watchdog Observer + """ + pass + + # ------------------------------------------------------------------------- + # --- Concrete methods providing Observer management --- + # ------------------------------------------------------------------------- + + def _setup_observer(self): + """Initialize the watchdog observer and common configuration.""" + self.observer = Observer() + source_config = self.source_location + self.watch_path = source_config.get('path') + if not self.watch_path: + raise ValueError("[parser.init.source] must contain a 'path' key.") + + self.recursive = source_config.get('recursive', True) + self.info(f"Observer setup for path: {self.watch_path}, recursive: {self.recursive}") + + def _start_observer(self): + """Start the watchdog observer with the event handler.""" + if not hasattr(self, 'observer') or not self.observer: + self._setup_observer() + + event_handler = self._create_event_handler() + self.observer.schedule(event_handler, self.watch_path, recursive=self.recursive) + self.observer.start() + self.info(f"Started {self.listener_name} observer on {self.watch_path}") + + def _stop_observer(self): + """Stop and join the watchdog observer.""" + if hasattr(self, 'observer') and self.observer: + self.observer.stop() + self.observer.join() + self.info(f"Stopped {self.listener_name} observer") + + def _run_observer_loop(self): + """Run the main observer loop with proper cleanup.""" + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + self.warning("Listener stopped by user. Processing any remaining files...") + self._process_batch() + self.end_notify(0) + finally: + self._stop_observer() + # ------------------------------------------------------------------------- # --- Concrete methods providing the batching functionality --- # ------------------------------------------------------------------------- diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py index 4601b508..43a7dcde 100644 --- a/source/listeners/watchdog_listener.py +++ b/source/listeners/watchdog_listener.py @@ -2,7 +2,6 @@ import time import threading from datetime import datetime -from watchdog.observers import Observer from watchdog.events import ( FileSystemEventHandler, FileClosedEvent, @@ -53,12 +52,16 @@ def listener_name(self) -> str: def listen(self): """Initializes batching and starts the watchdog observer with the regex handler.""" - self._setup_batching() - source_config = self.source_location - path = self.source_location.get('path') - if not path: - raise ValueError("[parser.init.source] must contain a 'path' key.") + self._setup_batching() + self._setup_observer() + self._start_observer() + self.start_notify() + self._run_observer_loop() + def _create_event_handler(self): + """Create the RegexMatchingEventHandler for this listener.""" + source_config = self.source_location + # Preparing arguments for the RegexMatchingEventHandler from the config file. handler_kwargs = { 'regexes': source_config.get('include_patterns', ['.*']), # Default to match everything @@ -66,24 +69,8 @@ def listen(self): 'ignore_directories': True, # Using the handler's built-in directory ignoring - helps empty directories 'case_sensitive': False } - - self.info(f"Starting file system listener on directory: {path}") - event_handler = PipelineEventHandler(parser_instance=self, **handler_kwargs) - observer = Observer() - observer.schedule(event_handler, path, recursive=True) - observer.start() - self.start_notify() - - try: - while True: - time.sleep(1) - except KeyboardInterrupt: - self.warning("Listener stopped by user. Processing any remaining files...") - self._process_batch() - self.end_notify(0) - finally: - observer.stop() - observer.join() + + return PipelineEventHandler(parser_instance=self, **handler_kwargs) def _parse_event(self, raw_event: FileSystemEventHandler) -> tuple[str | None, str | None]: """ From d757cf405a7427f5ff6ca3653ed6dac975d2116d Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Fri, 10 Oct 2025 20:12:48 -0500 Subject: [PATCH 03/11] Integrated listener to the existing framework --- configs/watchdog_listener_config.toml | 9 ++--- listen.py | 21 ++++++---- run.py | 28 ++++++++++++- source/common.py | 10 +++++ source/listeners/base.py | 57 +++++++++++++++++++-------- source/listeners/watchdog_listener.py | 8 ---- 6 files changed, 93 insertions(+), 40 deletions(-) diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml index 92e27b77..e16f049e 100644 --- a/configs/watchdog_listener_config.toml +++ b/configs/watchdog_listener_config.toml @@ -15,12 +15,11 @@ parts = [ [parser.init] # The directory where the state file ('upload_state.json') will be stored. -state_path = "C:/Users/Nihil/datanet_files/upload_state.json" - +state_path = "path/to/state/file" [parser.init.source] # The local directory that the listener will monitor for new files. -path = "E:/local" +path = "path/to/local/dir" #File name patterns to include/exclude. include_patterns = [".*\\.csv$", ".*\\.txt$"] exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] @@ -33,7 +32,7 @@ batch_max_latency_seconds = 60 [parser.init.target] # Settings for the S3 uploader. -bucket = "data-net-source-debug" +bucket = "bucket-name" # Your S3 bucket name prefix = "raw_data/" # Set to 'true' to upload files that have been modified. allow-overwrite = true @@ -42,7 +41,7 @@ allow-overwrite = true # Enable logging to a local file using the python's built-in logging. [logging.file] -filepath = "C:/Users/Nihil/datanet_files/s3_parser_log.txt" +filepath = "path/to/log/file.txt" level = 0 # level of events to include in this logger max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out max_files = 5 # Max number of files to keep before deleting the oldest diff --git a/listen.py b/listen.py index 49a33612..e07eb990 100644 --- a/listen.py +++ b/listen.py @@ -1,5 +1,6 @@ """ -Entry-point script to load and run a parser in continuous, real-time listener mode. +Entry-point script to run a parser in continuous listener mode. +This is equivalent to: python run.py config.toml --mode listen """ import argparse @@ -33,12 +34,16 @@ # 5. Call the .listen() method to start the long-running, event-driven process. try: parser.listen() - except AttributeError: - # Provide a helpful error if the user tries to run a checker-based config - # with this script, as it will not have a .listen() method. - parser.error( - f"The configured parser '{parser.__class__.__name__}' does not have a .listen() method. " - f"Ensure your config file specifies a Listener mixin, not a Checker mixin." - ) + except AttributeError as e: + if 'listen' in str(e): + # Provide a helpful error if the user tries to run a checker-based config + # with this script, as it will not have a .listen() method. + parser.error( + f"The configured parser '{parser.__class__.__name__}' does not have listener capabilities. " + f"This config appears to use checker mixins instead of listener mixins. " + f"Try using 'python run.py {args.config_file}' for checker mode." + ) + else: + parser.error(f"Parser setup failed: {e}", exc_info=True) except Exception as e: parser.error(f"The listener exited with an unexpected error: {e}", exc_info=True) \ No newline at end of file diff --git a/run.py b/run.py index dbddaf40..d219b58c 100644 --- a/run.py +++ b/run.py @@ -9,7 +9,7 @@ from os import PathLike -def load_config(config_fp: [str, PathLike]) -> dict: +def load_config(config_fp: str | PathLike) -> dict: with open(config_fp, 'r') as f: toml_config = toml.load(f) return toml_config @@ -54,6 +54,8 @@ def load_parser(parser_config: dict) -> ParserCommon: arg_parser = argparse.ArgumentParser() arg_parser.add_argument('config_file', type=str, help='Path to the config file that specifies the parser to run') + arg_parser.add_argument('--mode', choices=['checker', 'listen'], required=True, + help='Execution mode: checker (batch processing) or listen (continuous monitoring)') args = arg_parser.parse_args() config = load_config(args.config_file) @@ -61,4 +63,26 @@ def load_parser(parser_config: dict) -> ParserCommon: if 'logging' in config: parser.make_loggers(config['logging']) - parser.process() + # Execute in the specified mode + if args.mode == 'listen': + try: + parser.listen() + except AttributeError: + parser.error( + f"Parser '{parser.__class__.__name__}' does not support listen mode. " + f"Ensure your config uses listener mixins, not checker mixins." + ) + except Exception as e: + parser.error(f"Listen mode failed: {e}", exc_info=True) + else: + # Checker mode + try: + parser.process() + except AttributeError as e: + parser.error( + f"Parser '{parser.__class__.__name__}' does not support checker mode. " + f"This config appears to use listener mixins instead of checker mixins. " + f"Try using --mode listen instead." + ) + except Exception as e: + parser.error(f"Checker mode failed: {e}", exc_info=True) diff --git a/source/common.py b/source/common.py index 31e22d25..37913f9d 100644 --- a/source/common.py +++ b/source/common.py @@ -64,6 +64,7 @@ def __init__(self, state_path, source=None, middle=None, target=None): self.target_location = target def process(self): + self.start_notify() try: to_do = self.check() @@ -88,6 +89,15 @@ def process(self): else: self.end_notify(0) + def listen(self): + """ + Execute the parser in continuous mode - monitors for new data and processes it as it arrives. + This method should be implemented by listener mixins for event-driven behavior. + """ + # This will be overridden by listener mixins + # If called directly, it will cause an AttributeError that run.py will catch + pass + def check(self): """ Check should look for new data that needs to be uploaded diff --git a/source/listeners/base.py b/source/listeners/base.py index 880fb58b..44cb09df 100644 --- a/source/listeners/base.py +++ b/source/listeners/base.py @@ -7,18 +7,18 @@ from abc import ABC, abstractmethod from watchdog.observers import Observer -from source.common import EMPTY_LOG +from source.common import EMPTY_LOG, ParserCommon -class BaseListener(ABC): +class BaseListener(ParserCommon): """ - Abstract base class for all listener mixins. + Abstract base class for all listener mixins that extends ParserCommon. - This class defines the required interface for a listener component, thread-safe batching and provides - the common logic for processing events and managing state. + This class provides event-driven data processing capabilities with batching, + threading, and state management. It integrates seamlessly with the existing + checker/transformer/uploader pipeline by overriding the listen() method. """ state_filename = 'upload_state.json' - state_path = "path/to/state/save/dir" # ------------------------------------------------------------------------- # --- Abstract methods to be implemented by concrete listeners --- @@ -30,28 +30,51 @@ def listener_name(self) -> str: """A simple attribute naming the mixin class for later reference.""" pass - @abstractmethod def listen(self): """ - Start the listener. + Override ParserCommon's listen() method to provide event-driven processing. + + This method sets up the observer, starts monitoring, and processes events + as they arrive through the batching system. + """ + self.info(f"Starting {self.listener_name} in continuous mode...") + try: + self._setup_batching() + self._setup_observer() + self._start_observer() + self.start_notify() + self._run_observer_loop() + except KeyboardInterrupt: + self.warning("Listener stopped by user. Processing any remaining files...") + self._process_batch() + self.end_notify(0) + finally: + self._stop_observer() - This should be a blocking call that runs continuously to monitor an - event source (e.g., start a watchdog observer, connect to a message queue). + def check(self): """ - pass + Override ParserCommon's check() method to prevent using listener configs in checker mode. + + Listeners are designed for event-driven processing, not batch checking. + """ + raise AttributeError( + f"Listener '{self.listener_name}' does not support checker mode. " + f"This config uses listener mixins. Use --mode listen instead." + ) @abstractmethod - def _parse_event(self, raw_event: any) -> dict: + def _parse_event(self, raw_event: any) -> tuple[str | None, str | None]: """ - Parse a raw event from the source into the standard task format. + Parse a raw event from the source to extract file paths. - This is the primary abstraction point. It translates a source-specific - event object into the dictionary format expected by the `transform` method. + This method translates a source-specific event object into file paths + that can be processed by the batch system. :param raw_event: The native event object from the source (e.g., a watchdog event, a message from a queue). - :return: A dictionary in the format {'to do': [...], 'failure': [...]}. - Return an empty or None dict to indicate the event should be skipped. + :return: A tuple of (final_path, old_path). For simple events, old_path + should be None. For move events, both paths should be provided. + Return (None, None) to indicate the event should be skipped. """ pass diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py index 43a7dcde..55c47c58 100644 --- a/source/listeners/watchdog_listener.py +++ b/source/listeners/watchdog_listener.py @@ -50,14 +50,6 @@ class WatchdogListenerMixin(BaseListener): def listener_name(self) -> str: return "WatchdogBatchListener" - def listen(self): - """Initializes batching and starts the watchdog observer with the regex handler.""" - self._setup_batching() - self._setup_observer() - self._start_observer() - self.start_notify() - self._run_observer_loop() - def _create_event_handler(self): """Create the RegexMatchingEventHandler for this listener.""" source_config = self.source_location From e97b16388837f37f357b975d51839443aa4a09cd Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Wed, 22 Oct 2025 00:44:27 -0500 Subject: [PATCH 04/11] Abstracted base.py listener from ABC class and refactored code --- .gitignore | 1 + configs/watchdog_listener_config.toml | 8 +-- source/common.py | 36 ++++++++++-- source/listeners/base.py | 83 +++++++-------------------- source/listeners/watchdog_listener.py | 4 +- 5 files changed, 60 insertions(+), 72 deletions(-) diff --git a/.gitignore b/.gitignore index 04cd4519..35594af8 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ __pycache__/ requirements.txt env/ +configs/ diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml index e16f049e..bc0bb163 100644 --- a/configs/watchdog_listener_config.toml +++ b/configs/watchdog_listener_config.toml @@ -15,11 +15,11 @@ parts = [ [parser.init] # The directory where the state file ('upload_state.json') will be stored. -state_path = "path/to/state/file" +state_path = "C:/Users/Nihil/datanet_files/s3_upload_state.json" [parser.init.source] # The local directory that the listener will monitor for new files. -path = "path/to/local/dir" +path = "E:/local" #File name patterns to include/exclude. include_patterns = [".*\\.csv$", ".*\\.txt$"] exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] @@ -32,7 +32,7 @@ batch_max_latency_seconds = 60 [parser.init.target] # Settings for the S3 uploader. -bucket = "bucket-name" # Your S3 bucket name +bucket = "data-net-source-debug" # Your S3 bucket name prefix = "raw_data/" # Set to 'true' to upload files that have been modified. allow-overwrite = true @@ -41,7 +41,7 @@ allow-overwrite = true # Enable logging to a local file using the python's built-in logging. [logging.file] -filepath = "path/to/log/file.txt" +filepath = "C:/Users/Nihil/datanet_files/parser_log.txt" level = 0 # level of events to include in this logger max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out max_files = 5 # Max number of files to keep before deleting the oldest diff --git a/source/common.py b/source/common.py index 37913f9d..4d8434fa 100644 --- a/source/common.py +++ b/source/common.py @@ -91,12 +91,38 @@ def process(self): def listen(self): """ - Execute the parser in continuous mode - monitors for new data and processes it as it arrives. - This method should be implemented by listener mixins for event-driven behavior. + Execute the parser in continuous listener mode. + This is for event-driven, continuous operation. + + This orchestrates the listener lifecycle. """ - # This will be overridden by listener mixins - # If called directly, it will cause an AttributeError that run.py will catch - pass + self.start_notify() + try: + # Setup phase - prepare batching and observer + self.setup_batching() + self.setup_observer() + self.start_observer() + + # Main listening loop - runs continuously + self.info(f"Listener is now monitoring for events...") + self.run_observer_loop() + + except KeyboardInterrupt: + self.warning("Listener stopped by user. Processing any remaining files...") + # Process any remaining files in the batch + if hasattr(self, '_process_batch'): + self.process_batch() + self.end_notify(0) + except Exception as e: + import sys, traceback + self.error(e) + self.error(f'Encountered {str(e)}') + self.debug(traceback.format_exception(*sys.exc_info())) + self.end_notify(1) + finally: + # Cleanup phase - stop observer + if hasattr(self, 'stop_observer'): + self.stop_observer() def check(self): """ diff --git a/source/listeners/base.py b/source/listeners/base.py index 44cb09df..3bde6032 100644 --- a/source/listeners/base.py +++ b/source/listeners/base.py @@ -7,16 +7,16 @@ from abc import ABC, abstractmethod from watchdog.observers import Observer -from source.common import EMPTY_LOG, ParserCommon +from source.common import EMPTY_LOG -class BaseListener(ParserCommon): +class BaseListener(ABC): """ - Abstract base class for all listener mixins that extends ParserCommon. + Abstract base class for all listener mixins. This class provides event-driven data processing capabilities with batching, - threading, and state management. It integrates seamlessly with the existing - checker/transformer/uploader pipeline by overriding the listen() method. + threading, and state management. Listener mixins should be combined with + ParserCommon through dynamic class composition, similar to BaseChecker. """ state_filename = 'upload_state.json' @@ -30,40 +30,8 @@ def listener_name(self) -> str: """A simple attribute naming the mixin class for later reference.""" pass - def listen(self): - """ - Override ParserCommon's listen() method to provide event-driven processing. - - This method sets up the observer, starts monitoring, and processes events - as they arrive through the batching system. - """ - self.info(f"Starting {self.listener_name} in continuous mode...") - try: - self._setup_batching() - self._setup_observer() - self._start_observer() - self.start_notify() - self._run_observer_loop() - except KeyboardInterrupt: - self.warning("Listener stopped by user. Processing any remaining files...") - self._process_batch() - self.end_notify(0) - finally: - self._stop_observer() - - def check(self): - """ - Override ParserCommon's check() method to prevent using listener configs in checker mode. - - Listeners are designed for event-driven processing, not batch checking. - """ - raise AttributeError( - f"Listener '{self.listener_name}' does not support checker mode. " - f"This config uses listener mixins. Use --mode listen instead." - ) - @abstractmethod - def _parse_event(self, raw_event: any) -> tuple[str | None, str | None]: + def parse_event(self, raw_event: any) -> tuple[str | None, str | None]: """ Parse a raw event from the source to extract file paths. @@ -93,7 +61,7 @@ def clean(self): pass @abstractmethod - def _create_event_handler(self): + def create_event_handler(self): """ Create the specific event handler for this listener type. @@ -105,7 +73,7 @@ def _create_event_handler(self): # --- Concrete methods providing Observer management --- # ------------------------------------------------------------------------- - def _setup_observer(self): + def setup_observer(self): """Initialize the watchdog observer and common configuration.""" self.observer = Observer() source_config = self.source_location @@ -116,40 +84,33 @@ def _setup_observer(self): self.recursive = source_config.get('recursive', True) self.info(f"Observer setup for path: {self.watch_path}, recursive: {self.recursive}") - def _start_observer(self): + def start_observer(self): """Start the watchdog observer with the event handler.""" if not hasattr(self, 'observer') or not self.observer: - self._setup_observer() - - event_handler = self._create_event_handler() + self.setup_observer() + + event_handler = self.create_event_handler() self.observer.schedule(event_handler, self.watch_path, recursive=self.recursive) self.observer.start() self.info(f"Started {self.listener_name} observer on {self.watch_path}") - def _stop_observer(self): + def stop_observer(self): """Stop and join the watchdog observer.""" if hasattr(self, 'observer') and self.observer: self.observer.stop() self.observer.join() self.info(f"Stopped {self.listener_name} observer") - def _run_observer_loop(self): - """Run the main observer loop with proper cleanup.""" - try: - while True: - time.sleep(1) - except KeyboardInterrupt: - self.warning("Listener stopped by user. Processing any remaining files...") - self._process_batch() - self.end_notify(0) - finally: - self._stop_observer() + def run_observer_loop(self): + """Run the main observer loop. Cleanup is handled by ParserCommon.listen().""" + while True: + time.sleep(1) # ------------------------------------------------------------------------- # --- Concrete methods providing the batching functionality --- # ------------------------------------------------------------------------- - def _setup_batching(self): + def setup_batching(self): """Initializes all common batching attributes from the configuration.""" source_config = self.source_location self._batch_max_size = source_config.get('batch_max_size', 10) @@ -165,7 +126,7 @@ def add_to_batch(self, raw_event: any): Adds a file from an event to the batch buffer, handles renames, and checks batch triggers using a unified logic. """ - final_path, old_path = self._parse_event(raw_event) + final_path, old_path = self.parse_event(raw_event) if not final_path: return # The event was ignored by the parser. @@ -191,7 +152,7 @@ def add_to_batch(self, raw_event: any): # Start the timer only if this is the first item in a new batch. if len(self._file_buffer) == 1: self._batch_timer = threading.Timer( - self._batch_max_latency_seconds, self._process_batch + self._batch_max_latency_seconds, self.process_batch ) self._batch_timer.start() @@ -200,9 +161,9 @@ def add_to_batch(self, raw_event: any): process_now = True if process_now: - self._process_batch() + self.process_batch() - def _process_batch(self): + def process_batch(self): """Processes all files currently in the buffer.""" files_to_process = [] with self._buffer_lock: diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py index 55c47c58..38b3d018 100644 --- a/source/listeners/watchdog_listener.py +++ b/source/listeners/watchdog_listener.py @@ -50,7 +50,7 @@ class WatchdogListenerMixin(BaseListener): def listener_name(self) -> str: return "WatchdogBatchListener" - def _create_event_handler(self): + def create_event_handler(self): """Create the RegexMatchingEventHandler for this listener.""" source_config = self.source_location @@ -64,7 +64,7 @@ def _create_event_handler(self): return PipelineEventHandler(parser_instance=self, **handler_kwargs) - def _parse_event(self, raw_event: FileSystemEventHandler) -> tuple[str | None, str | None]: + def parse_event(self, raw_event: FileSystemEventHandler) -> tuple[str | None, str | None]: """ Parses a watchdog event to extract file paths and filter unwanted files, returning a (final_path, old_path) tuple. Path-based filtering is From a723c0979004af3c11de4bdceca14d5698229ea8 Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Wed, 22 Oct 2025 00:47:34 -0500 Subject: [PATCH 05/11] Adding config generator script --- configs/generate_config.py | 418 +++++++++++++++++++++++++++++++++++++ 1 file changed, 418 insertions(+) create mode 100644 configs/generate_config.py diff --git a/configs/generate_config.py b/configs/generate_config.py new file mode 100644 index 00000000..b65a81f5 --- /dev/null +++ b/configs/generate_config.py @@ -0,0 +1,418 @@ +#!/usr/bin/env python3 +""" +Interactive Config Generator for Data Net Source + +This script helps users create configuration files by selecting from available +checkers, listeners, transformers, and uploaders. +""" + +import os +import sys +from pathlib import Path + + +# Available components registry +CHECKERS = { + "1": { + "name": "FileCheckerMixin", + "module": "checkers.local.__init__", + "description": "Check local directory for new files", + "source_config": { + "path": "path/to/local/directory", + "check_for_modifications": False, + }, + "dependencies": [] + }, + "2": { + "name": "StreamedFileCheckerMixin", + "module": "checkers.local.stream", + "description": "Check local directory for streamed files", + "source_config": { + "path": "path/to/local/directory", + "streamed_files": "*", + "stream_rate": 60, + "reliability_factor": 1.0, + }, + "dependencies": [] + }, + "3": { + "name": "S3CheckerMixin", + "module": "checkers.s3", + "description": "Check AWS S3 bucket for new files", + "source_config": { + "bucket": "source-bucket-name", + "prefix": "path/in/bucket/", + }, + "dependencies": ["boto3"] + }, + "4": { + "name": "OuraAPICheckerMixin", + "module": "checkers.api.oura", + "description": "Fetch data from Oura Ring API", + "source_config": { + "path": "path/to/store/downloaded/data", + "oura_config": "path/to/oura/config.json", + }, + "dependencies": ["requests"] + }, + "5": { + "name": "QualtricsAPICheckerMixin", + "module": "checkers.api.qualtrics", + "description": "Fetch data from Qualtrics API", + "source_config": { + "path": "path/to/store/downloaded/data", + "qualtrics_config": "path/to/qualtrics/config.json", + }, + "dependencies": ["requests"] + }, + "6": { + "name": "REDCapAPICheckerMixin", + "module": "checkers.api.redcap", + "description": "Fetch data from REDCap API", + "source_config": { + "path": "path/to/store/downloaded/data", + "redcap_config": "path/to/redcap/config.json", + }, + "dependencies": ["requests"] + }, + "7": { + "name": "RuneAPICheckerMixin", + "module": "checkers.api.rune", + "description": "Fetch data from Rune Labs API", + "source_config": { + "path": "path/to/store/downloaded/data", + "rune_config": "path/to/rune/config", + "rune_patients_config": "path/to/patients/config.json", + }, + "dependencies": ["runeq"] + }, +} + +LISTENERS = { + "1": { + "name": "WatchdogListenerMixin", + "module": "listeners.watchdog_listener", + "description": "Monitor filesystem for new files in real-time", + "source_config": { + "path": "path/to/monitor", + "include_patterns": [".*\\.csv$", ".*\\.txt$"], + "exclude_patterns": [".*\\.tmp$"], + "batch_max_size": 10, + "batch_max_latency_seconds": 60, + }, + "dependencies": ["watchdog"] + }, +} + +TRANSFORMERS = { + "0": { + "name": "None", + "module": None, + "description": "No transformation (pass-through)", + "middle_config": None, + "dependencies": [] + }, + "1": { + "name": "OpenPoseTransformer", + "module": "transformers.openpose", + "description": "Process videos with OpenPose for pose estimation", + "middle_config": { + "path": "path/to/intermediate/storage", + }, + "dependencies": [] + }, + "2": { + "name": "MatlabTransformerMixin", + "module": "transformers.matlab", + "description": "Process data using MATLAB scripts", + "middle_config": { + "path": "path/to/intermediate/storage", + }, + "dependencies": [] + }, +} + +UPLOADERS = { + "1": { + "name": "S3UploaderMixin", + "module": "uploaders.local_to_s3", + "description": "Upload files to AWS S3 bucket", + "target_config": { + "bucket": "target-bucket-name", + "prefix": "raw_data/", + "allow-overwrite": True, + }, + "dependencies": ["boto3"] + }, + "2": { + "name": "CopyUploaderMixin", + "module": "uploaders.simple", + "description": "Copy files to local directory", + "target_config": { + "path": "path/to/destination", + }, + "dependencies": [] + }, + "3": { + "name": "FTPUploaderMixin", + "module": "uploaders.ftp", + "description": "Upload files via FTP", + "target_config": { + "host": "ftp.example.com", + "port": 21, + "username": "username", + "password": "password", + "path": "/remote/path", + }, + "dependencies": [] + }, + "4": { + "name": "SSHUploaderMixin", + "module": "uploaders.ssh", + "description": "Upload files via SSH/SFTP", + "target_config": { + "host": "ssh.example.com", + "port": 22, + "username": "username", + "key_file": "path/to/ssh/key", + "path": "/remote/path", + }, + "dependencies": ["paramiko"] + }, +} + + +def print_header(text): + """Print a formatted header.""" + print("\n" + "=" * 70) + print(f" {text}") + print("=" * 70) + + +def print_options(options_dict, title): + """Print available options.""" + print(f"\n{title}:") + print("-" * 70) + for key, value in options_dict.items(): + print(f" [{key}] {value['name']}") + print(f" {value['description']}") + + +def get_user_choice(prompt, valid_choices): + """Get and validate user input.""" + while True: + choice = input(f"\n{prompt}: ").strip() + if choice in valid_choices: + return choice + print(f"Invalid choice. Please select from: {', '.join(valid_choices)}") + + +def generate_toml_config(parser_name, mode, checker_or_listener, transformer, uploader, state_path, log_path): + """Generate TOML configuration content.""" + + # Determine which component to use + if mode == "checker": + primary_component = CHECKERS[checker_or_listener] + else: + primary_component = LISTENERS[checker_or_listener] + + transformer_component = TRANSFORMERS[transformer] + uploader_component = UPLOADERS[uploader] + + # Build parts list + parts = [] + parts.append([primary_component["module"], primary_component["name"]]) + + if transformer_component["module"]: + parts.append([transformer_component["module"], transformer_component["name"]]) + + parts.append([uploader_component["module"], uploader_component["name"]]) + + # Collect dependencies + dependencies = set() + dependencies.update(primary_component["dependencies"]) + dependencies.update(transformer_component["dependencies"]) + dependencies.update(uploader_component["dependencies"]) + + # Start building TOML content + config_lines = [] + config_lines.append("# " + "=" * 60) + config_lines.append(f"# Auto-generated config for {parser_name}") + config_lines.append(f"# Mode: {mode}") + config_lines.append("# " + "=" * 60) + config_lines.append("") + config_lines.append("[parser]") + config_lines.append("") + config_lines.append("[parser.class]") + config_lines.append('type = "dynamic"') + config_lines.append(f'name = "{parser_name}"') + config_lines.append("parts = [") + for module, name in parts: + config_lines.append(f' ["{module}", "{name}"],') + config_lines.append("]") + config_lines.append("") + + # Parser initialization + config_lines.append("[parser.init]") + config_lines.append(f'state_path = "{state_path}"') + config_lines.append("") + + # Source configuration + config_lines.append("[parser.init.source]") + for key, value in primary_component["source_config"].items(): + if isinstance(value, str): + config_lines.append(f'{key} = "{value}"') + elif isinstance(value, bool): + config_lines.append(f'{key} = {str(value).lower()}') + elif isinstance(value, list): + config_lines.append(f'{key} = {value}') + else: + config_lines.append(f'{key} = {value}') + config_lines.append("") + + # Middle configuration (if transformer is used) + if transformer_component["middle_config"]: + config_lines.append("[parser.init.middle]") + for key, value in transformer_component["middle_config"].items(): + config_lines.append(f'{key} = "{value}"') + config_lines.append("") + + # Target configuration + config_lines.append("[parser.init.target]") + for key, value in uploader_component["target_config"].items(): + if isinstance(value, str): + config_lines.append(f'{key} = "{value}"') + elif isinstance(value, bool): + config_lines.append(f'{key} = {str(value).lower()}') + else: + config_lines.append(f'{key} = {value}') + config_lines.append("") + + # Logging configuration + config_lines.append("# Logging configuration") + config_lines.append("[logging.file]") + config_lines.append(f'filepath = "{log_path}"') + config_lines.append("level = 0 # 0=DEBUG, 10=INFO, 20=WARNING, 30=ERROR") + config_lines.append("max_size = 21048576 # 20MB") + config_lines.append("max_files = 5") + config_lines.append("") + + # Dependencies + if dependencies: + config_lines.append("[dependencies]") + config_lines.append("pip = [") + for dep in sorted(dependencies): + config_lines.append(f' "{dep}",') + config_lines.append("]") + + return "\n".join(config_lines) + + +def main(): + """Main interactive configuration generator.""" + print_header("Data Net Source - Config Generator") + print("\nThis tool will help you create a configuration file for your parser.") + + # Step 1: Choose mode + print_header("Step 1: Choose Parser Mode") + print("\nSelect the execution mode for your parser:") + print(" [1] Checker Mode - Batch processing (runs once per execution)") + print(" [2] Listen Mode - Continuous monitoring (event-driven)") + + mode_choice = get_user_choice("Select mode [1/2]", ["1", "2"]) + mode = "checker" if mode_choice == "1" else "listen" + + # Step 2: Choose checker or listener + if mode == "checker": + print_header("Step 2: Choose Checker") + print_options(CHECKERS, "Available Checkers") + checker_choice = get_user_choice( + f"Select checker [1-{len(CHECKERS)}]", + list(CHECKERS.keys()) + ) + primary_choice = checker_choice + else: + print_header("Step 2: Choose Listener") + print_options(LISTENERS, "Available Listeners") + listener_choice = get_user_choice( + f"Select listener [1-{len(LISTENERS)}]", + list(LISTENERS.keys()) + ) + primary_choice = listener_choice + + # Step 3: Choose transformer (optional) + print_header("Step 3: Choose Transformer (Optional)") + print_options(TRANSFORMERS, "Available Transformers") + transformer_choice = get_user_choice( + f"Select transformer [0-{len(TRANSFORMERS)-1}]", + list(TRANSFORMERS.keys()) + ) + + # Step 4: Choose uploader + print_header("Step 4: Choose Uploader") + print_options(UPLOADERS, "Available Uploaders") + uploader_choice = get_user_choice( + f"Select uploader [1-{len(UPLOADERS)}]", + list(UPLOADERS.keys()) + ) + + # Step 5: Basic configuration + print_header("Step 5: Basic Configuration") + parser_name = input("\nEnter parser name (e.g., MyDataParser): ").strip() + if not parser_name: + parser_name = "CustomParser" + + state_path = input("Enter state directory path: ").strip() + if not state_path: + state_path = "path/to/state/directory" + + log_path = input("Enter log file path: ").strip() + if not log_path: + log_path = "path/to/log/file.log" + + # Generate config + print_header("Generating Configuration") + config_content = generate_toml_config( + parser_name, mode, primary_choice, transformer_choice, + uploader_choice, state_path, log_path + ) + + # Save config + output_filename = input("\nEnter output filename (e.g., my_parser.toml): ").strip() + if not output_filename: + output_filename = "generated_config.toml" + + if not output_filename.endswith('.toml'): + output_filename += '.toml' + + # Determine output path (generated_configs directory) + script_dir = Path(__file__).parent + generated_dir = script_dir / "generated_configs" + + # Create generated_configs directory if it doesn't exist + generated_dir.mkdir(exist_ok=True) + + output_path = generated_dir / output_filename + + with open(output_path, 'w') as f: + f.write(config_content) + + print_header("Configuration Generated Successfully!") + print(f"\nConfig file saved to: {output_path}") + print("\nNext steps:") + print(f" 1. Review and edit the config file: {output_filename}") + print(f" 2. Replace placeholder values with your actual paths and settings") + print(f" 3. Run: python prepare.py {output_filename}") + print(f" 4. Run: python run.py {output_filename} --mode {mode}") + print("\n" + "=" * 70) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\n\nConfig generation cancelled.") + sys.exit(0) + except Exception as e: + print(f"\n\nError: {e}") + sys.exit(1) From 3768505e646ed64655c245f8e40dfbdbdb253ae3 Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Wed, 22 Oct 2025 00:51:06 -0500 Subject: [PATCH 06/11] removed file paths --- configs/watchdog_listener_config.toml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml index bc0bb163..f4d243db 100644 --- a/configs/watchdog_listener_config.toml +++ b/configs/watchdog_listener_config.toml @@ -15,11 +15,11 @@ parts = [ [parser.init] # The directory where the state file ('upload_state.json') will be stored. -state_path = "C:/Users/Nihil/datanet_files/s3_upload_state.json" +state_path = "path/to/state/file" [parser.init.source] # The local directory that the listener will monitor for new files. -path = "E:/local" +path = "path/to/local/directory" #File name patterns to include/exclude. include_patterns = [".*\\.csv$", ".*\\.txt$"] exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] @@ -32,7 +32,7 @@ batch_max_latency_seconds = 60 [parser.init.target] # Settings for the S3 uploader. -bucket = "data-net-source-debug" # Your S3 bucket name +bucket = "target-bucket-name" # Your S3 bucket name prefix = "raw_data/" # Set to 'true' to upload files that have been modified. allow-overwrite = true @@ -41,7 +41,7 @@ allow-overwrite = true # Enable logging to a local file using the python's built-in logging. [logging.file] -filepath = "C:/Users/Nihil/datanet_files/parser_log.txt" +filepath = "path/to/log/file.log" level = 0 # level of events to include in this logger max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out max_files = 5 # Max number of files to keep before deleting the oldest From fc8c4a28e9a8e37341aa5f3fe6362872e8c99a3c Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Wed, 22 Oct 2025 00:52:13 -0500 Subject: [PATCH 07/11] Added watchdog_config to ignore --- .gitignore | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 35594af8..33c89a15 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,4 @@ __pycache__/ requirements.txt env/ -configs/ +configs/watchdog_listener_config.py From 8d737c04288d51bd699f0466b41fccbef929ae8a Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Tue, 28 Oct 2025 16:10:31 -0500 Subject: [PATCH 08/11] removing config generator --- configs/generate_config.py | 418 -------------------------- configs/watchdog_listener_config.toml | 8 +- 2 files changed, 4 insertions(+), 422 deletions(-) delete mode 100644 configs/generate_config.py diff --git a/configs/generate_config.py b/configs/generate_config.py deleted file mode 100644 index b65a81f5..00000000 --- a/configs/generate_config.py +++ /dev/null @@ -1,418 +0,0 @@ -#!/usr/bin/env python3 -""" -Interactive Config Generator for Data Net Source - -This script helps users create configuration files by selecting from available -checkers, listeners, transformers, and uploaders. -""" - -import os -import sys -from pathlib import Path - - -# Available components registry -CHECKERS = { - "1": { - "name": "FileCheckerMixin", - "module": "checkers.local.__init__", - "description": "Check local directory for new files", - "source_config": { - "path": "path/to/local/directory", - "check_for_modifications": False, - }, - "dependencies": [] - }, - "2": { - "name": "StreamedFileCheckerMixin", - "module": "checkers.local.stream", - "description": "Check local directory for streamed files", - "source_config": { - "path": "path/to/local/directory", - "streamed_files": "*", - "stream_rate": 60, - "reliability_factor": 1.0, - }, - "dependencies": [] - }, - "3": { - "name": "S3CheckerMixin", - "module": "checkers.s3", - "description": "Check AWS S3 bucket for new files", - "source_config": { - "bucket": "source-bucket-name", - "prefix": "path/in/bucket/", - }, - "dependencies": ["boto3"] - }, - "4": { - "name": "OuraAPICheckerMixin", - "module": "checkers.api.oura", - "description": "Fetch data from Oura Ring API", - "source_config": { - "path": "path/to/store/downloaded/data", - "oura_config": "path/to/oura/config.json", - }, - "dependencies": ["requests"] - }, - "5": { - "name": "QualtricsAPICheckerMixin", - "module": "checkers.api.qualtrics", - "description": "Fetch data from Qualtrics API", - "source_config": { - "path": "path/to/store/downloaded/data", - "qualtrics_config": "path/to/qualtrics/config.json", - }, - "dependencies": ["requests"] - }, - "6": { - "name": "REDCapAPICheckerMixin", - "module": "checkers.api.redcap", - "description": "Fetch data from REDCap API", - "source_config": { - "path": "path/to/store/downloaded/data", - "redcap_config": "path/to/redcap/config.json", - }, - "dependencies": ["requests"] - }, - "7": { - "name": "RuneAPICheckerMixin", - "module": "checkers.api.rune", - "description": "Fetch data from Rune Labs API", - "source_config": { - "path": "path/to/store/downloaded/data", - "rune_config": "path/to/rune/config", - "rune_patients_config": "path/to/patients/config.json", - }, - "dependencies": ["runeq"] - }, -} - -LISTENERS = { - "1": { - "name": "WatchdogListenerMixin", - "module": "listeners.watchdog_listener", - "description": "Monitor filesystem for new files in real-time", - "source_config": { - "path": "path/to/monitor", - "include_patterns": [".*\\.csv$", ".*\\.txt$"], - "exclude_patterns": [".*\\.tmp$"], - "batch_max_size": 10, - "batch_max_latency_seconds": 60, - }, - "dependencies": ["watchdog"] - }, -} - -TRANSFORMERS = { - "0": { - "name": "None", - "module": None, - "description": "No transformation (pass-through)", - "middle_config": None, - "dependencies": [] - }, - "1": { - "name": "OpenPoseTransformer", - "module": "transformers.openpose", - "description": "Process videos with OpenPose for pose estimation", - "middle_config": { - "path": "path/to/intermediate/storage", - }, - "dependencies": [] - }, - "2": { - "name": "MatlabTransformerMixin", - "module": "transformers.matlab", - "description": "Process data using MATLAB scripts", - "middle_config": { - "path": "path/to/intermediate/storage", - }, - "dependencies": [] - }, -} - -UPLOADERS = { - "1": { - "name": "S3UploaderMixin", - "module": "uploaders.local_to_s3", - "description": "Upload files to AWS S3 bucket", - "target_config": { - "bucket": "target-bucket-name", - "prefix": "raw_data/", - "allow-overwrite": True, - }, - "dependencies": ["boto3"] - }, - "2": { - "name": "CopyUploaderMixin", - "module": "uploaders.simple", - "description": "Copy files to local directory", - "target_config": { - "path": "path/to/destination", - }, - "dependencies": [] - }, - "3": { - "name": "FTPUploaderMixin", - "module": "uploaders.ftp", - "description": "Upload files via FTP", - "target_config": { - "host": "ftp.example.com", - "port": 21, - "username": "username", - "password": "password", - "path": "/remote/path", - }, - "dependencies": [] - }, - "4": { - "name": "SSHUploaderMixin", - "module": "uploaders.ssh", - "description": "Upload files via SSH/SFTP", - "target_config": { - "host": "ssh.example.com", - "port": 22, - "username": "username", - "key_file": "path/to/ssh/key", - "path": "/remote/path", - }, - "dependencies": ["paramiko"] - }, -} - - -def print_header(text): - """Print a formatted header.""" - print("\n" + "=" * 70) - print(f" {text}") - print("=" * 70) - - -def print_options(options_dict, title): - """Print available options.""" - print(f"\n{title}:") - print("-" * 70) - for key, value in options_dict.items(): - print(f" [{key}] {value['name']}") - print(f" {value['description']}") - - -def get_user_choice(prompt, valid_choices): - """Get and validate user input.""" - while True: - choice = input(f"\n{prompt}: ").strip() - if choice in valid_choices: - return choice - print(f"Invalid choice. Please select from: {', '.join(valid_choices)}") - - -def generate_toml_config(parser_name, mode, checker_or_listener, transformer, uploader, state_path, log_path): - """Generate TOML configuration content.""" - - # Determine which component to use - if mode == "checker": - primary_component = CHECKERS[checker_or_listener] - else: - primary_component = LISTENERS[checker_or_listener] - - transformer_component = TRANSFORMERS[transformer] - uploader_component = UPLOADERS[uploader] - - # Build parts list - parts = [] - parts.append([primary_component["module"], primary_component["name"]]) - - if transformer_component["module"]: - parts.append([transformer_component["module"], transformer_component["name"]]) - - parts.append([uploader_component["module"], uploader_component["name"]]) - - # Collect dependencies - dependencies = set() - dependencies.update(primary_component["dependencies"]) - dependencies.update(transformer_component["dependencies"]) - dependencies.update(uploader_component["dependencies"]) - - # Start building TOML content - config_lines = [] - config_lines.append("# " + "=" * 60) - config_lines.append(f"# Auto-generated config for {parser_name}") - config_lines.append(f"# Mode: {mode}") - config_lines.append("# " + "=" * 60) - config_lines.append("") - config_lines.append("[parser]") - config_lines.append("") - config_lines.append("[parser.class]") - config_lines.append('type = "dynamic"') - config_lines.append(f'name = "{parser_name}"') - config_lines.append("parts = [") - for module, name in parts: - config_lines.append(f' ["{module}", "{name}"],') - config_lines.append("]") - config_lines.append("") - - # Parser initialization - config_lines.append("[parser.init]") - config_lines.append(f'state_path = "{state_path}"') - config_lines.append("") - - # Source configuration - config_lines.append("[parser.init.source]") - for key, value in primary_component["source_config"].items(): - if isinstance(value, str): - config_lines.append(f'{key} = "{value}"') - elif isinstance(value, bool): - config_lines.append(f'{key} = {str(value).lower()}') - elif isinstance(value, list): - config_lines.append(f'{key} = {value}') - else: - config_lines.append(f'{key} = {value}') - config_lines.append("") - - # Middle configuration (if transformer is used) - if transformer_component["middle_config"]: - config_lines.append("[parser.init.middle]") - for key, value in transformer_component["middle_config"].items(): - config_lines.append(f'{key} = "{value}"') - config_lines.append("") - - # Target configuration - config_lines.append("[parser.init.target]") - for key, value in uploader_component["target_config"].items(): - if isinstance(value, str): - config_lines.append(f'{key} = "{value}"') - elif isinstance(value, bool): - config_lines.append(f'{key} = {str(value).lower()}') - else: - config_lines.append(f'{key} = {value}') - config_lines.append("") - - # Logging configuration - config_lines.append("# Logging configuration") - config_lines.append("[logging.file]") - config_lines.append(f'filepath = "{log_path}"') - config_lines.append("level = 0 # 0=DEBUG, 10=INFO, 20=WARNING, 30=ERROR") - config_lines.append("max_size = 21048576 # 20MB") - config_lines.append("max_files = 5") - config_lines.append("") - - # Dependencies - if dependencies: - config_lines.append("[dependencies]") - config_lines.append("pip = [") - for dep in sorted(dependencies): - config_lines.append(f' "{dep}",') - config_lines.append("]") - - return "\n".join(config_lines) - - -def main(): - """Main interactive configuration generator.""" - print_header("Data Net Source - Config Generator") - print("\nThis tool will help you create a configuration file for your parser.") - - # Step 1: Choose mode - print_header("Step 1: Choose Parser Mode") - print("\nSelect the execution mode for your parser:") - print(" [1] Checker Mode - Batch processing (runs once per execution)") - print(" [2] Listen Mode - Continuous monitoring (event-driven)") - - mode_choice = get_user_choice("Select mode [1/2]", ["1", "2"]) - mode = "checker" if mode_choice == "1" else "listen" - - # Step 2: Choose checker or listener - if mode == "checker": - print_header("Step 2: Choose Checker") - print_options(CHECKERS, "Available Checkers") - checker_choice = get_user_choice( - f"Select checker [1-{len(CHECKERS)}]", - list(CHECKERS.keys()) - ) - primary_choice = checker_choice - else: - print_header("Step 2: Choose Listener") - print_options(LISTENERS, "Available Listeners") - listener_choice = get_user_choice( - f"Select listener [1-{len(LISTENERS)}]", - list(LISTENERS.keys()) - ) - primary_choice = listener_choice - - # Step 3: Choose transformer (optional) - print_header("Step 3: Choose Transformer (Optional)") - print_options(TRANSFORMERS, "Available Transformers") - transformer_choice = get_user_choice( - f"Select transformer [0-{len(TRANSFORMERS)-1}]", - list(TRANSFORMERS.keys()) - ) - - # Step 4: Choose uploader - print_header("Step 4: Choose Uploader") - print_options(UPLOADERS, "Available Uploaders") - uploader_choice = get_user_choice( - f"Select uploader [1-{len(UPLOADERS)}]", - list(UPLOADERS.keys()) - ) - - # Step 5: Basic configuration - print_header("Step 5: Basic Configuration") - parser_name = input("\nEnter parser name (e.g., MyDataParser): ").strip() - if not parser_name: - parser_name = "CustomParser" - - state_path = input("Enter state directory path: ").strip() - if not state_path: - state_path = "path/to/state/directory" - - log_path = input("Enter log file path: ").strip() - if not log_path: - log_path = "path/to/log/file.log" - - # Generate config - print_header("Generating Configuration") - config_content = generate_toml_config( - parser_name, mode, primary_choice, transformer_choice, - uploader_choice, state_path, log_path - ) - - # Save config - output_filename = input("\nEnter output filename (e.g., my_parser.toml): ").strip() - if not output_filename: - output_filename = "generated_config.toml" - - if not output_filename.endswith('.toml'): - output_filename += '.toml' - - # Determine output path (generated_configs directory) - script_dir = Path(__file__).parent - generated_dir = script_dir / "generated_configs" - - # Create generated_configs directory if it doesn't exist - generated_dir.mkdir(exist_ok=True) - - output_path = generated_dir / output_filename - - with open(output_path, 'w') as f: - f.write(config_content) - - print_header("Configuration Generated Successfully!") - print(f"\nConfig file saved to: {output_path}") - print("\nNext steps:") - print(f" 1. Review and edit the config file: {output_filename}") - print(f" 2. Replace placeholder values with your actual paths and settings") - print(f" 3. Run: python prepare.py {output_filename}") - print(f" 4. Run: python run.py {output_filename} --mode {mode}") - print("\n" + "=" * 70) - - -if __name__ == "__main__": - try: - main() - except KeyboardInterrupt: - print("\n\nConfig generation cancelled.") - sys.exit(0) - except Exception as e: - print(f"\n\nError: {e}") - sys.exit(1) diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml index f4d243db..bc0bb163 100644 --- a/configs/watchdog_listener_config.toml +++ b/configs/watchdog_listener_config.toml @@ -15,11 +15,11 @@ parts = [ [parser.init] # The directory where the state file ('upload_state.json') will be stored. -state_path = "path/to/state/file" +state_path = "C:/Users/Nihil/datanet_files/s3_upload_state.json" [parser.init.source] # The local directory that the listener will monitor for new files. -path = "path/to/local/directory" +path = "E:/local" #File name patterns to include/exclude. include_patterns = [".*\\.csv$", ".*\\.txt$"] exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] @@ -32,7 +32,7 @@ batch_max_latency_seconds = 60 [parser.init.target] # Settings for the S3 uploader. -bucket = "target-bucket-name" # Your S3 bucket name +bucket = "data-net-source-debug" # Your S3 bucket name prefix = "raw_data/" # Set to 'true' to upload files that have been modified. allow-overwrite = true @@ -41,7 +41,7 @@ allow-overwrite = true # Enable logging to a local file using the python's built-in logging. [logging.file] -filepath = "path/to/log/file.log" +filepath = "C:/Users/Nihil/datanet_files/parser_log.txt" level = 0 # level of events to include in this logger max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out max_files = 5 # Max number of files to keep before deleting the oldest From 4a9084037fe75455d7d09099b86ec4d93cf2b125 Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Tue, 28 Oct 2025 16:13:42 -0500 Subject: [PATCH 09/11] Added watchdog_config file with nameholders --- configs/watchdog_listener_config.toml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml index bc0bb163..e16f049e 100644 --- a/configs/watchdog_listener_config.toml +++ b/configs/watchdog_listener_config.toml @@ -15,11 +15,11 @@ parts = [ [parser.init] # The directory where the state file ('upload_state.json') will be stored. -state_path = "C:/Users/Nihil/datanet_files/s3_upload_state.json" +state_path = "path/to/state/file" [parser.init.source] # The local directory that the listener will monitor for new files. -path = "E:/local" +path = "path/to/local/dir" #File name patterns to include/exclude. include_patterns = [".*\\.csv$", ".*\\.txt$"] exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"] @@ -32,7 +32,7 @@ batch_max_latency_seconds = 60 [parser.init.target] # Settings for the S3 uploader. -bucket = "data-net-source-debug" # Your S3 bucket name +bucket = "bucket-name" # Your S3 bucket name prefix = "raw_data/" # Set to 'true' to upload files that have been modified. allow-overwrite = true @@ -41,7 +41,7 @@ allow-overwrite = true # Enable logging to a local file using the python's built-in logging. [logging.file] -filepath = "C:/Users/Nihil/datanet_files/parser_log.txt" +filepath = "path/to/log/file.txt" level = 0 # level of events to include in this logger max_size = 21048576 # Max size, in bytes, of a log file before it is rotated out max_files = 5 # Max number of files to keep before deleting the oldest From 701ff2d8b831d5d8793db28643125057abc94f42 Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Thu, 19 Mar 2026 14:44:34 -0500 Subject: [PATCH 10/11] Refactor argument parser in run.py and now run.py handles only checker mode and listen.py listener mode --- run.py | 42 +++++++++++++++--------------------------- 1 file changed, 15 insertions(+), 27 deletions(-) diff --git a/run.py b/run.py index d219b58c..3bb78c30 100644 --- a/run.py +++ b/run.py @@ -51,11 +51,11 @@ def load_parser(parser_config: dict) -> ParserCommon: if __name__ == '__main__': - arg_parser = argparse.ArgumentParser() + arg_parser = argparse.ArgumentParser( + description="Run a data-net-source parser in batch checker mode." + ) arg_parser.add_argument('config_file', type=str, - help='Path to the config file that specifies the parser to run') - arg_parser.add_argument('--mode', choices=['checker', 'listen'], required=True, - help='Execution mode: checker (batch processing) or listen (continuous monitoring)') + help='Path to the .toml config file that specifies the parser to run') args = arg_parser.parse_args() config = load_config(args.config_file) @@ -63,26 +63,14 @@ def load_parser(parser_config: dict) -> ParserCommon: if 'logging' in config: parser.make_loggers(config['logging']) - # Execute in the specified mode - if args.mode == 'listen': - try: - parser.listen() - except AttributeError: - parser.error( - f"Parser '{parser.__class__.__name__}' does not support listen mode. " - f"Ensure your config uses listener mixins, not checker mixins." - ) - except Exception as e: - parser.error(f"Listen mode failed: {e}", exc_info=True) - else: - # Checker mode - try: - parser.process() - except AttributeError as e: - parser.error( - f"Parser '{parser.__class__.__name__}' does not support checker mode. " - f"This config appears to use listener mixins instead of checker mixins. " - f"Try using --mode listen instead." - ) - except Exception as e: - parser.error(f"Checker mode failed: {e}", exc_info=True) + # Checker mode (batch processing) + try: + parser.process() + except AttributeError as e: + parser.error( + f"Parser '{parser.__class__.__name__}' does not support checker mode. " + f"This config appears to use listener mixins instead of checker mixins. " + f"Try using 'python listen.py {args.config_file}' for continuous listener mode." + ) + except Exception as e: + parser.error(f"Checker mode failed: {e}", exc_info=True) From 63de8b32ab78a3114ab97c5e970eb61065c8b990 Mon Sep 17 00:00:00 2001 From: NikhilDS-Rice Date: Thu, 19 Mar 2026 14:48:36 -0500 Subject: [PATCH 11/11] Refactor state handling in WatchdogListenerMixin to process completed events more efficiently --- source/listeners/watchdog_listener.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py index 38b3d018..0c124645 100644 --- a/source/listeners/watchdog_listener.py +++ b/source/listeners/watchdog_listener.py @@ -91,9 +91,9 @@ def save(self, completed: dict): event['timestamp'] = now current_state = self.load_state() - for key in ['success', 'failure', 'skipped']: - if key in completed and completed[key]: - current_state[key].extend(completed[key]) + for key, events in completed.items(): + if events: + current_state[key].extend(events) self.write_state(current_state) def clean(self):