Merge pull request #175 from CHESSComputing/foxden-writers

vkuznet · web-flow · commit 130b2fbf03da · 2025-03-16T09:26:06.000-04:00
Add implementation for FoxdenMetaDataWriter and FoxdenMetaDataProcessor
diff --git a/CHAP/foxden/__init__.py b/CHAP/foxden/__init__.py
@@ -5,4 +5,4 @@
         FoxdenProvenanceProcessor, FoxdenMetaDataProcessor
 from CHAP.foxden.reader import \
         FoxdenMetaDataReader, FoxdenProvenanceReader, FoxdenSpecScansReader
-from CHAP.foxden.writer import FoxdenWriter
+from CHAP.foxden.writer import FoxdenProvenanceWriter, FoxdenMetaDataWriter
diff --git a/CHAP/foxden/processor.py b/CHAP/foxden/processor.py
@@ -9,59 +9,168 @@
 
 # System modules
 from time import time
+import os
+import sys
+import platform
+import pkg_resources
+import subprocess
+import json
 
 # Local modules
 from CHAP.processor import Processor
 
 class FoxdenMetaDataProcessor(Processor):
     """A Processor to communicate with FOXDEN MetaData server."""
 
-    def process(self, data, url, did, dryRun=False, verbose=False):
+    def process(self, data, suffix='analysis=CHAP', verbose=False):
         """FOXDEN MetaData processor
 
         :param data: Input data.
         :type data: list[PipelineData]
-        :param url: URL of service.
-        :type url: str
-        :param did: FOXDEN dataset identifier (did)
-        :type did: string
-        :param dryRun: `dryRun` option to verify HTTP workflow,
-            defaults to `False`.
-        :type dryRun: bool, optional
+        :param suffix: did suffix to add, default 'analysis=CHAP'
+        :type suffix: string, optional
         :param verbose: verbose output
         :type verbose: bool, optional
         :return: data from FOXDEN MetaData service
         """
         t0 = time()
         self.logger.info(
-            f'Executing "process" with url={url} data={data} did={did}')
+            f'Executing "process" with data={data}')
+        output = []
+        for item in data:
+            # each item in data list is a CHAP record {'name': ..., 'data': {}}
+            for rec in item['data']:  # get data part of processing item
+                if 'did' not in rec:
+                    raise Exception('No did found in input data record')
+                did = rec['did'] + '/' + suffix
+                # construct analysis record
+                rec = {'did': did, 'application': 'CHAP'}
+                output.append(rec)
         self.logger.info(f'Finished "process" in {time()-t0:.3f} seconds\n')
-        return data
+        return output
 
 class FoxdenProvenanceProcessor(Processor):
     """A Processor to communicate with FOXDEN provenance server."""
-    def process(self, data, url, did, dryRun=False, verbose=False):
+    def process(self, data, suffix='analysis=CHAP', verbose=False):
         """FOXDEN Provenance processor
 
         :param data: Input data.
         :type data: list[PipelineData]
-        :param url: URL of service.
-        :type url: str
-        :param did: FOXDEN dataset identifier (did)
-        :type did: string
-        :param dryRun: `dryRun` option to verify HTTP workflow,
-            defaults to `False`.
-        :type dryRun: bool, optional
+        :param suffix: did suffix to add, default 'analysis=CHAP'
+        :type suffix: string, optional
         :param verbose: verbose output
         :type verbose: bool, optional
         :return: data from FOXDEN provenance service
         """
         t0 = time()
         self.logger.info(
-            f'Executing "process" with url={url} data={data} did={did}')
+            f'Executing "process" with data={data}')
+        output = []
+        for item in data:
+            # each item in data list is a CHAP record {'name': ..., 'data': {}}
+            for rec in item['data']:  # get data part of processing item
+                if 'did' not in rec:
+                    raise Exception('No did found in input data record')
+                rec['did'] = rec['did'] + '/' + suffix
+                rec['parent_did'] = rec['did'] 
+                rec['scripts'] = [{'name': 'CHAP', 'parent_script': None, 'order_idx': 1}]
+                rec['site'] = 'Cornell'
+                rec['osinfo'] = osinfo()
+                rec['environments'] = environments()
+                rec['input_files'] = inputFiles()
+                rec['output_files'] = outputFiles()
+                rec['processing'] = 'CHAP pipeline'
+                output.append(rec)
         self.logger.info(f'Finished "process" in {time()-t0:.3f} seconds\n')
-        return data
+        return output
 
+def osinfo():
+    """
+    Helper function to provide osinfo
+    """
+    os_info = {
+        "name": platform.system().lower() + "-" + platform.release(),
+        "kernel": platform.version(),
+        "version": platform.platform()
+    }
+    return os_info
+
+def environments():
+    """
+    Detects the current Python environment (system, virtualenv, or Conda) and
+    collects package information. Returns a list of detected environments with
+    installed packages.
+    """
+    environments = []
+    os_name = platform.system().lower() + "-" + platform.release()
+
+    # Check for Conda environment
+    conda_env = os.getenv("CONDA_PREFIX")
+    if conda_env:
+        conda_env_name = os.getenv("CONDA_DEFAULT_ENV", "unknown-conda-env")
+        try:
+            # Fetch Conda packages
+            conda_packages = subprocess.check_output(["conda", "list", "--json"], text=True)
+            conda_packages = json.loads(conda_packages)
+            packages = [{"name": pkg["name"], "version": pkg["version"]} for pkg in conda_packages]
+        except Exception:
+            packages = []
+
+        environments.append({
+            "name": conda_env_name,
+            "version": sys.version.split()[0],
+            "details": "Conda environment",
+            "parent_environment": None,
+            "os_name": os_name,
+            "packages": packages
+        })
+
+    # Check for Virtualenv (excluding Conda)
+    elif hasattr(sys, 'real_prefix') or os.getenv("VIRTUAL_ENV"):
+        venv_name = os.path.basename(os.getenv("VIRTUAL_ENV", "unknown-venv"))
+        packages = [
+            {"name": pkg.key, "version": pkg.version}
+            for pkg in pkg_resources.working_set
+        ]
+
+        environments.append({
+            "name": venv_name,
+            "version": sys.version.split()[0],
+            "details": "Virtualenv environment",
+            "parent_environment": None,
+            "os_name": os_name,
+            "packages": packages
+        })
+
+    # System Python (not inside Conda or Virtualenv)
+    else:
+        packages = [
+            {"name": pkg.key, "version": pkg.version}
+            for pkg in pkg_resources.working_set
+        ]
+
+        environments.append({
+            "name": "system-python",
+            "version": sys.version.split()[0],
+            "details": "System-wide Python",
+            "parent_environment": None,
+            "os_name": os_name,
+            "packages": packages
+        })
+
+    return environments
+
+def inputFiles():
+    """
+    Helper function to provide input files for FOXDEN
+    """
+    return [{'name':'/tmp/file1.png'}, {'name': '/tmp/file2.png'}]
+
+def outputFiles():
+    """
+    Helper function to provide output files for FOXDEN
+    """
+    return [{'name':'/tmp/file1.png'}]
 
 if __name__ == '__main__':
     # Local modules
diff --git a/CHAP/foxden/reader.py b/CHAP/foxden/reader.py
@@ -24,7 +24,7 @@ class FoxdenMetaDataReader(PipelineItem):
     def read(
             self, url, data, did='', query='', spec=None,
             method='GET', headers=None,
-            scope='read', timeout=10, dryRun=False, verbose=False):
+            scope='read', dryRun=False, verbose=False):
         """Read data from FOXDEN service
 
         :param url: URL of service.
@@ -44,8 +44,6 @@ def read(
         :type headers: dictionary, optional
         :param scope: FOXDEN scope to use, e.g. read or write
         :type scope: string
-        :param timeout: Timeout of HTTP request, defaults to `10`.
-        :type timeout: str, optional
         :param dryRun: `dryRun` option to verify HTTP workflow,
             defaults to `False`.
         :type dryRun: bool, optional
@@ -82,7 +80,7 @@ class FoxdenProvenanceReader(PipelineItem):
     """FOXDEN Provenance reader reads data from specific FOXDEN Provenance service."""
     def read(
             self, url, data, did='', method='GET', headers=None,
-            scope='read', timeout=10, dryRun=False, verbose=False):
+            scope='read', dryRun=False, verbose=False):
         """FOXDEN Provenance processor
 
         :param url: URL of service.
@@ -128,7 +126,7 @@ class FoxdenSpecScansReader(PipelineItem):
     def read(
             self, url, data, did='', query='', spec=None,
             method='GET', headers=None,
-            scope='read', timeout=10, dryRun=False, verbose=False):
+            scope='read', dryRun=False, verbose=False):
         """Read data from FOXDEN service
 
         :param url: URL of service.
@@ -148,8 +146,6 @@ def read(
         :type headers: dictionary, optional
         :param scope: FOXDEN scope to use, e.g. read or write
         :type scope: string
-        :param timeout: Timeout of HTTP request, defaults to `10`.
-        :type timeout: str, optional
         :param dryRun: `dryRun` option to verify HTTP workflow,
             defaults to `False`.
         :type dryRun: bool, optional
diff --git a/CHAP/foxden/writer.py b/CHAP/foxden/writer.py
@@ -1,14 +1,65 @@
-"""FOXDEN writer."""
+#!/usr/bin/env python
+#-*- coding: utf-8 -*-
+#pylint: disable=
+"""
+File       : writer.py
+Author     : Valentin Kuznetsov <vkuznet AT gmail dot com>
+Description: FOXDEN writers
+"""
 
+# system modules
+from time import time
+import json
+
+# CHAP modules
 from CHAP.foxden.utils import HttpRequest
+from CHAP.pipeline import PipelineItem
+
+
+class FoxdenMetaDataWriter(PipelineItem):
+    """FOXDEN writer writes data to MetaData FOXDEN service."""
+    def write(
+            self, url, data, method='POST', headers=None, verbose=False):
+        """Write data to FOXDEN Provenance service
 
+        :param data: Input data.
+        :type data: list[PipelineData]
+        :param url: URL of service.
+        :type url: str
+        :param method: HTTP method to use, `"POST"` for creation and
+            `"PUT"` for update, defaults to `"POST"`.
+        :type method: str, optional
+        :param headers: HTTP headers to use.
+        :type headers: dictionary, optional
+        :param verbose: verbose output
+        :type verbose: bool, optional
+        :return: HTTP response from FOXDEN provenance service
+        :rtype: list with dictionary entry
+        """
+        t0 = time()
+        self.logger.info(
+            f'Executing "process" with url={url} data={data}')
+        # TODO: it would be useful to perform validation of data
+        if isinstance(data, list) and len(data) == 1:
+            data = data[0]['data'][0]
+        if not isinstance(data, dict):
+            raise Exception(f'Passed data={data} is not dictionary')
+        mrec = {"Schema": "Analysis", "Record": data}
+        payload = json.dumps(mrec)
+        if verbose:
+            self.logger.info(f"method=POST url={url} payload={payload}")
+        response = HttpRequest(url, payload, method='POST', scope='write')
+        if verbose:
+            self.logger.info(f"code={response.status_code} data={response.text}")
+        data = [{'code': response.status_code, 'data': response.text}]
+        self.logger.info(f'Finished "process" in {time()-t0:.3f} seconds\n')
+        return data
 
-class FoxdenWriter():
-    """FOXDEN writer writes data to specific FOXDEN service."""
+class FoxdenProvenanceWriter(PipelineItem):
+    """FOXDEN writer writes data to provenance FOXDEN service."""
     def write(
-            self, url, data, method='POST', headers=None,
-            scope='write', timeout=10, dryRun=False):
-        """Write data to FOXDEN
+            self, url, data, method='POST', headers=None, verbose=False):
+        """Write data to FOXDEN Provenance service
 
         :param data: Input data.
         :type data: list[PipelineData]
@@ -18,18 +69,30 @@ def write(
             `"PUT"` for update, defaults to `"POST"`.
         :type method: str, optional
         :param headers: HTTP headers to use.
-        :param scope: FOXDEN scope to use, e.g. read or write
-        :type scope: string
         :type headers: dictionary, optional
-        :param timeout: Timeout of HTTP request, defaults to `10`.
-        :type timeout: str, optional
-        :param dryRun: `dryRun` option to verify HTTP workflow,
-            defaults to `False`.
-        :type dryRun: bool, optional
-        :return: Contents of the input data.
-        :rtype: object
+        :param verbose: verbose output
+        :type verbose: bool, optional
+        :return: HTTP response from FOXDEN provenance service
+        :rtype: list with dictionary entry
         """
-        return HttpRequest(url, data, method, headers, scope, timeout, dryrun)
+        t0 = time()
+        self.logger.info(
+            f'Executing "process" with url={url} data={data}')
+        rurl = f'{url}/dataset'
+        # TODO: it would be useful to perform validation of data
+        if isinstance(data, list) and len(data) == 1:
+            data = data[0]['data'][0]
+        if not isinstance(data, dict):
+            raise Exception(f'Passed data={data} is not dictionary')
+        payload = json.dumps(data)
+        if verbose:
+            self.logger.info(f"method=POST url={rurl} payload={payload}")
+        response = HttpRequest(rurl, payload, method='POST', scope='write')
+        if verbose:
+            self.logger.info(f"code={response.status_code} data={response.text}")
+        data = [{'code': response.status_code, 'data': response.text}]
+        self.logger.info(f'Finished "process" in {time()-t0:.3f} seconds\n')
+        return data
 
 
 if __name__ == '__main__':
