Adding Extensions to Support Main Pipeline

.github/workflows/docker-release.yml → .github/workflows/docker-build.yml

@@ -1,10 +1,12 @@
-name: Docker Image CI
+name: Build Docker Image
 
 on:
   push:
-    branches: [ "main" ]
+    branches: 
+      - '*'
   pull_request:
-    branches: [ "main" ]
+    branches: 
+      - '*'
 
 jobs:
 
@@ -16,3 +18,4 @@ jobs:
     - uses: actions/checkout@v4
     - name: Build the Docker image
       run: docker build . --file Dockerfile --tag my-image-name:$(date +%s)
+

.github/workflows/main.yaml

@@ -18,8 +18,19 @@ jobs:
       uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
-    - name: Install echodataflow
-      run: pip install .[all]
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pytest pytest-cov
+        pip install -r requirements.txt
+        pip install -e .
+
+    - name: Initialize
+      run: |
+        echodataflow init
+        prefect profiles create echodataflow-local
+
     - name: Run unit tests
       env:
         FORCE_COLOR: 3

deployment/Linux/FileDownloader.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=Prefect serve
+After=network.target
+
+[Service]
+User=exouser
+WorkingDirectory=/home/exouser/Desktop/Echodataflow
+ExecStart=/home/exouser/Desktop/echodataflowv2/bin/python /home/exouser/Desktop/echodataflowv2/lib/python3.10/site-packages/echodataflow/extensions/file_downloader.py
+Restart=always
+StandardOutput=syslog
+StandardError=syslog
+SyslogIdentifier=echodataflow
+
+[Install]
+WantedBy=multi-user.target

deployment/Linux/FileMonitor.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=Prefect serve
+After=network.target
+
+[Service]
+User=exouser
+WorkingDirectory=/home/exouser/Desktop/Echodataflow
+ExecStart=/home/exouser/Desktop/echodataflowv2/bin/python /home/exouser/Desktop/echodataflowv2/lib/python3.10/site-packages/echodataflow/extensions/file_monitor.py
+Restart=always
+StandardOutput=syslog
+StandardError=syslog
+SyslogIdentifier=echodataflow
+
+[Install]
+WantedBy=multi-user.target

deployment/Linux/echodataflow.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=Prefect serve
+After=network.target
+
+[Service]
+User=exouser
+WorkingDirectory=/home/exouser/Desktop/Echodataflow/
+ExecStart=/home/exouser/Desktop/echodataflowv2/bin/python /home/exouser/Desktop/Echodataflow/main.py
+Restart=always
+StandardOutput=syslog
+StandardError=syslog
+SyslogIdentifier=echodataflow
+
+[Install]
+WantedBy=multi-user.target

deployment/MacOS/FileDownloader.plist

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+  <dict>
+    <key>Label</key>
+    <string>com.example.filedownloader</string>
+    <key>UserName</key>
+    <string>exouser</string>
+    <key>WorkingDirectory</key>
+    <string>/Users/exouser/Desktop/Echodataflow</string>
+    <key>ProgramArguments</key>
+    <array>
+      <string>/Users/exouser/Desktop/echodataflowv2/bin/python</string>
+      <string>/Users/exouser/Desktop/echodataflowv2/lib/python3.10/site-packages/echodataflow/extensions/file_downloader.py</string>
+    </array>
+    <key>RunAtLoad</key>
+    <true/>
+    <key>KeepAlive</key>
+    <true/>
+    <key>StandardOutPath</key>
+    <string>/var/log/filedownloader.log</string>
+    <key>StandardErrorPath</key>
+    <string>/var/log/filedownloader.err</string>
+  </dict>
+</plist>

deployment/MacOS/FileMonitor.plist

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+  <dict>
+    <key>Label</key>
+    <string>com.example.filemonitor</string>
+    <key>UserName</key>
+    <string>exouser</string>
+    <key>WorkingDirectory</key>
+    <string>/Users/exouser/Desktop/Echodataflow</string>
+    <key>ProgramArguments</key>
+    <array>
+      <string>/Users/exouser/Desktop/echodataflowv2/bin/python</string>
+      <string>/Users/exouser/Desktop/echodataflowv2/lib/python3.10/site-packages/echodataflow/extensions/file_monitor.py</string>
+    </array>
+    <key>RunAtLoad</key>
+    <true/>
+    <key>KeepAlive</key>
+    <true/>
+    <key>StandardOutPath</key>
+    <string>/var/log/filemonitor.log</string>
+    <key>StandardErrorPath</key>
+    <string>/var/log/filemonitor.err</string>
+  </dict>
+</plist>

deployment/MacOS/echodataflow.plist

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+  <dict>
+    <key>Label</key>
+    <string>com.example.echodataflow</string>
+    <key>UserName</key>
+    <string>exouser</string>
+    <key>WorkingDirectory</key>
+    <string>/Users/exouser/Desktop/Echodataflow/</string>
+    <key>ProgramArguments</key>
+    <array>
+      <string>/Users/exouser/Desktop/echodataflowv2/bin/python</string>
+      <string>/Users/exouser/Desktop/Echodataflow/main.py</string>
+    </array>
+    <key>RunAtLoad</key>
+    <true/>
+    <key>KeepAlive</key>
+    <true/>
+    <key>StandardOutPath</key>
+    <string>/var/log/echodataflow.log</string>
+    <key>StandardErrorPath</key>
+    <string>/var/log/echodataflow.err</string>
+  </dict>
+</plist>

deployment/docker_trigger.py

@@ -0,0 +1,56 @@
+import os
+import json
+from pathlib import Path
+from echoflow import echoflow_start
+from echoflow.stages.echoflow_trigger import echoflow_trigger
+from prefect import flow
+from prefect.task_runners import SequentialTaskRunner
+from typing import Any, Dict, Optional, Union
+
+@flow(name="Docker-Trigger", task_runner=SequentialTaskRunner())
+def docker_trigger(
+    dataset_config: Union[dict, str, Path],
+    pipeline_config: Union[dict, str, Path],
+    logging_config: Union[dict, str, Path] = None,
+    storage_options: Optional[dict] = None,
+    options: Optional[dict] = {},
+    json_data_path: Union[str, Path] = None
+):
+    return echoflow_trigger(
+        dataset_config=dataset_config,
+        pipeline_config=pipeline_config,
+        logging_config=logging_config,
+        storage_options=storage_options,
+        options=options,
+        json_data_path=json_data_path
+    )
+
+def convert_to_proper_type(input_str):
+    if input_str is None:
+        return None
+
+    # Try to parse input as JSON
+    try:
+        return json.loads(input_str)
+    except (ValueError, TypeError):
+        pass
+
+    # If it's not valid JSON, check if it's a file path
+    if os.path.exists(input_str):
+        return Path(input_str)
+
+    # If not JSON or a valid file path, return the input as is
+    return input_str
+
+if __name__ == "__main__":
+    print("Entered")
+    # Get inputs from environment variables or command-line arguments
+    dataset_config = convert_to_proper_type(os.getenv('DATASET_CONFIG', '{}'))
+    pipeline_config = convert_to_proper_type(os.getenv('PIPELINE_CONFIG', '{}'))
+    logging_config = convert_to_proper_type(os.getenv('LOGGING_CONFIG', None))
+    storage_options = convert_to_proper_type(os.getenv('STORAGE_OPTIONS', None))
+    options = convert_to_proper_type(os.getenv('OPTIONS', '{}'))
+    json_data_path = convert_to_proper_type(os.getenv('JSON_DATA_PATH', None))
+
+    # Call the docker_trigger function
+    docker_trigger(dataset_config, pipeline_config, logging_config, storage_options, options, json_data_path)

deployment/main.py

@@ -0,0 +1,28 @@
+import logging
+from echodataflow import echodataflow_start
+from prefect import flow
+from prefect.blocks.core import Block
+from typing import Any, Dict, Optional, Union
+from pathlib import Path
+
+logging.basicConfig(filename='/home/exouser/Desktop/Echodataflow/echodataflow/logs/service.log', level=logging.DEBUG, format='%(asctime)s %(levelname)s:%(message)s')
+
+@flow
+def echodataflow(dataset_config: Union[Dict[str, Any], str, Path],
+    pipeline_config: Union[Dict[str, Any], str, Path],
+    logging_config: Union[Dict[str, Any], str, Path] = None,
+    storage_options: Union[Dict[str, Any], Block] = None,
+    options: Optional[Dict[str, Any]] = {},
+    json_data_path: Union[str, Path] = None):
+
+    echodataflow_start(dataset_config=dataset_config, pipeline_config=pipeline_config, logging_config=logging_config
+                       , storage_options=storage_options, options=options, json_data_path=json_data_path)
+
+if __name__ == "__main__":
+    logging.debug("Starting the Echodataflow service")
+    try:
+        echodataflow.serve(name="Echodataflow")
+        logging.debug("Echodataflow service started successfully")
+    except Exception as e:
+        logging.error(f"Error starting the Echodataflow service: {e}")
+        raise

echodataflow/aspects/echodataflow_aspect.py

@@ -46,18 +46,18 @@ def my_function(arg1, arg2):
             # Function code here
             pass
     """
-
     def decorator(func=None):
         def before_function_call(
             gea: Singleton_Echodataflow, type: str, processing_stage: str, *args, **kwargs
         ):
+
             if gea:
                 gea.log(
                     msg=f"Entering with memory at {gea.log_memory_usage()}: ",
                     extra={"mod_name": func.__module__, "func_name": func.__name__},
                     level=logging.DEBUG,
                 )
-
+            
             # Deprecating, since we have check before starting the pipeline
 
             # if type == "FLOW" and processing_stage!= "DEFAULT":

echodataflow/aspects/singleton_echodataflow.py

@@ -55,20 +55,21 @@ class Singleton_Echodataflow:
     dataset: Dataset
     db_log: DB_Log
     logger: logging.Logger = None
-    log_level: int = 0
+    log_level: int  = 0
 
     def __new__(
         cls,
         log_file: Union[Dict[str, str], str] = None,
         pipeline: Recipe = None,
-        dataset: Dataset = None,
+        dataset: Dataset = None 
     ) -> "Singleton_Echodataflow":
+
         if cls._instance is None:
             cls._instance = super().__new__(cls)
             if log_file is not None:
                 cls._instance.logger = cls._instance.logger_init(log_file)
                 cls._instance.log_level = cls._instance.logger.level
-
+                
         cls._instance.pipeline = pipeline
         cls._instance.dataset = dataset
         # cls._instance.db_log = cls._instance.setup_echodataflow_db()
@@ -84,7 +85,7 @@ def get_instance(self) -> "Singleton_Echodataflow":
             Singleton_Echodataflow: The Singleton_Echodataflow instance.
         """
         return self._instance
-
+    
     @classmethod
     def get_logger(self) -> "logging.Logger":
         """
@@ -143,10 +144,7 @@ def log(self, msg, level, extra):
             extra: Extra information to include in the log record.
         """
         if self.logger is not None:
-            self.logger.log(level=logging.DEBUG, msg=msg, extra=extra)
-            self.logger.log(level=logging.ERROR, msg=msg, extra=extra)
-            self.logger.log(level=logging.INFO, msg=msg, extra=extra)
-            self.logger.log(level=logging.WARNING, msg=msg, extra=extra)
+            self.logger.log(level=level, msg=msg, extra=extra)            
         else:
             print(f"{extra} : {msg}")
 
@@ -182,6 +180,7 @@ def insert_log_data(self):
             print(e)
         finally:
             conn.close()
+
 
     def log_memory_usage(self):
         """
@@ -193,7 +192,7 @@ def log_memory_usage(self):
         process = psutil.Process()
         mem = process.memory_info().rss
         return mem
-
+    
     def get_possible_next_functions(self, function_name: str):
         """
         Get Possible Next Functions
@@ -212,7 +211,7 @@ def get_possible_next_functions(self, function_name: str):
             next_functions = executor.get_possible_next_functions(function_name="data_download")
         """
         return self.rengine.get_possible_next_functions(function_name)
-
+        
     def load(self):
         """
         Load Rules and Dependencies
@@ -228,12 +227,10 @@ def load(self):
             loaded_rengine = executor.load()
         """
         rengine: DependencyEngine = DependencyEngine()
-        rules_file_path = os.path.expanduser(
-            os.path.join("~", ".echodataflow", "echodataflow_rules.txt")
-        )
+        rules_file_path = os.path.expanduser(os.path.join("~", ".echodataflow", "echodataflow_rules.txt"))
 
-        with open(rules_file_path, "r") as file:
+        with open(rules_file_path, 'r') as file:
             for line in file:
-                target, dependent = line.strip("\r\n").split(":")
+                target, dependent = line.strip().split(':')
                 rengine.add_dependency(target_function=target, dependent_function=dependent)
         return rengine