Merge pull request #6 from odtp-org/output-results

v0.1.0

Author: caviri

.env.dist

@@ -0,0 +1,20 @@
+# ODTP ENV VARIABLES TO CONNECT
+ODTP_MONGO_SERVER=
+ODTP_S3_SERVER=
+ODTP_BUCKET_NAME=
+ODTP_ACCESS_KEY=
+ODTP_SECRET_KEY=
+
+# ODTP ENV VARIABLES DB REFERENCES
+ODTP_USER_ID=
+ODTP_DIGITAL_TWIN=
+ODTP_EXCUTION=
+ODTP_STEP=
+ODTP_COMPONENT=
+ODTP_COMPONENT_VERSION=
+ODTP_RESULT=
+
+# OTHER ENV VARIABLES (FALSE/TRUE)
+TRANSFER_INPUT_TO_OUTPUT=FALSE
+ODTP_SAVE_SNAPSHOT=FALSE
+ODTP_SAVE_IN_RESULT=TRUE

README.md

@@ -4,10 +4,15 @@ Client for ODTP Components. This repository is aimed to be used as a submodule i
 
 ## Changelog
 
+- v0.1.0
+    - Logging in `logs` collection as individual document. 
+    - Updating `output` document
+    - Updating `result` document
+
 - v0.0.1
     - First version
     - Inluding conditional to make snapshot optionals
-    - Adding quiet flag to compressing
+    - Adding quiet flag to compressing odtp-output
 
 # Acknowledgments, Copyright, and Licensing
 ## Acknowledgments and Funding

logger.py

@@ -2,7 +2,7 @@
 
 from pymongo import MongoClient, errors
 from bson import ObjectId
-from datetime import datetime
+from datetime import datetime, timezone
 import os
 import time
 import json
@@ -46,6 +46,13 @@ def add_digital_twin(self, userRef, digital_twin_data):
         )
 
         return digital_twin_id
+    
+    def add_logs(self, log_data_list):
+        logs_collection = self.db["logs"]
+
+        log_ids = logs_collection.insert_many(log_data_list).inserted_ids
+
+        return log_ids
 
 
     def append_execution(self, digital_twin_id, execution_data):
@@ -78,18 +85,18 @@ def append_step(self, execution_id, step_data):
         return step_id
 
     def add_output(self, step_id, output_data):
-        steps_collection = self.db["outputs"]
+        output_collection = self.db["outputs"]
         output_data["stepRef"] = step_id
 
         # TODO: Make its own function. Taking out user_id
         #output_data["access_control"]["authorized_users"] = user_id
 
-        output_id = steps_collection.insert_one(output_data).inserted_id
+        output_id = output_collection.insert_one(output_data).inserted_id
 
-        # Update digital twin with execution reference
-        self.db.digitalTwins.update_one(
+        # Update steps with execution reference
+        self.db.steps.update_one(
             {"_id": ObjectId(step_id)},  # Specify the document to update
-            {"$set": {"field_name": output_id}}  # Use $set to replace the value of a field
+            {"$set": {"output": output_id}}  # Use $set to replace the value of a field
         )
 
         return output_id
@@ -108,15 +115,25 @@ def append_logs(self, step_id, log_data_list):
             {"_id": ObjectId(step_id)},
             {"$push": {"logs": {"$each": log_data_list}}}
         )
-
-    def add_result(self, digital_twin_id, execution_id, result_data):
+    
+    def update_result(self, result_id, output_id):
         results_collection = self.db["results"]
-        result_data["digitalTwinRef"] = digital_twin_id
-        result_data["executionRed"] = execution_id
-        
-        result_id = results_collection.insert_one(result_data).inserted_id
+        results_collection.update_one(
+            {"_id": ObjectId(result_id)},
+            {"$push": {"output": output_id}}
+        )
+
+        results_collection.update_one(
+            {"_id": ObjectId(result_id)},
+            {"$set": {"updated_at": datetime.now(timezone.utc)}}
+        )
 
-        return result_id
+    def update_end_time(self, step_id):
+        steps_collection = self.db["steps"]
+        steps_collection.update_one(
+            {"_id": ObjectId(step_id)},
+            {"$set": {"end_timestamp": datetime.now(timezone.utc)}}
+        )
 
     ######### Get methods
 
@@ -260,21 +277,26 @@ def main(delay=2):
         newLogList = []
         for log in logs:
             newLogEntry= {
-            "timestamp": datetime.utcnow(),
-            "type": "INFO",
+            "stepRef": step_id,
+            "timestamp": datetime.now(timezone.utc),
             "logstring": log}
 
             newLogList.append(newLogEntry)
 
 
         dbManager = MongoManager(MONGO_URL, db_name)
-        _ = dbManager.append_logs(step_id, newLogList)
+        # _ = dbManager.append_logs(step_id, newLogList)
+        _ = dbManager.add_logs(newLogList)
         dbManager.close()
 
         time.sleep(0.2)
 
         # TODO: Improve this
         if log == "--- ODTP COMPONENT ENDING ---":
+            dbManager = MongoManager(MONGO_URL, db_name)
+            dbManager.update_end_time(step_id)
+            dbManager.close()
+
             ending_detected = True
 
         #time.sleep(delay)

s3uploader.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 import os
 import time
 import logging
@@ -176,8 +176,8 @@ def main():
         "file_name": "odtp-output.zip",  # The name of the file in the output
         "file_size": file_size_bytes,  # Size of the file in bytes
         "file_type": "application/zip",  # MIME type or file type
-        "created_at": datetime.utcnow(),  # Timestamp when the output was created
-        "updated_at": datetime.utcnow(),  # Timestamp when the output was last updated
+        "created_at": datetime.now(timezone.utc),  # Timestamp when the output was created
+        "updated_at": datetime.now(timezone.utc),  # Timestamp when the output was last updated
         "metadata": {  # Additional metadata associated with the output
             "description": "Description of the snapshot",
             "tags": ["tag1", "tag2"],
@@ -190,6 +190,9 @@ def main():
     }
 
     odtp_output_id = dbManager.add_output(STEP_ID, output_data)
+
+    if os.getenv("ODTP_SAVE_IN_RESULT") == "TRUE":
+        dbManager.update_result(os.getenv("ODTP_RESULT"), odtp_output_id)
 
     logging.info("ODTP OUTPUT UPLOADED IN {}".format(odtp_output_id))
 
@@ -209,8 +212,8 @@ def main():
             "file_name": "odtp-snapshot.zip",  # The name of the file in the output
             "file_size": file_size_bytes,  # Size of the file in bytes
             "file_type": "application/zip",  # MIME type or file type
-            "created_at": datetime.utcnow(),  # Timestamp when the output was created
-            "updated_at": datetime.utcnow(),  # Timestamp when the output was last updated
+            "created_at": datetime.now(timezone.utc),  # Timestamp when the output was created
+            "updated_at": datetime.now(timezone.utc),  # Timestamp when the output was last updated
             "metadata": {  # Additional metadata associated with the output
                 "description": "Description of the snapshot",
                 "tags": ["tag1", "tag2"],