SIENTIAPDE-994

Refactor activity preparation methods and update Docker configuration for simulator

laborious/activities/activities.py

@@ -50,5 +50,5 @@ class Activities(Postgres, MLFlow, Gates, OPC):
                      notification_handler=notification_handler)
 
     @activity.defn(name="prepare_activity")
-    async def prepare_activity(self, schedule_name: str, model_name: str, model_id: str):
-        await super().prepare_activity(schedule_name, model_name, model_id)
+    def prepare_activity(self, input_data: dict[str, Any]):
+        super().prepare_activity(input_data)

laborious/activities/base.py

@@ -1,6 +1,6 @@
+from logging import Logger
 from temporalio import activity
 from sientia_do.notifications.handlers import NotificationHandler
-from logging import Logger
 
 
 class BaseActivity:
@@ -8,17 +8,18 @@ class BaseActivity:
         self.logger = logger
         self.notification_handler = notification_handler
 
-    def prepare_activity(self, schedule_name: str,
-                         model_name: str,
-                         model_id: str):
+    @activity.defn(name="prepare_activity")
+    def prepare_activity(self, input_data: dict[str, Any]):
         """
         Prepare the activity for the notification handler.
 
         Args:
+            workflow_name (str): The name of the workflow.
             schedule_name (str): The name of the schedule.
             model_name (str): The name of the model.
             model_id (str): The id of the model.
         """
-        self.notification_handler.base_notification.schedule_name = schedule_name
-        self.notification_handler.base_notification.model_name = model_name
-        self.notification_handler.base_notification.model_id = model_id
+        self.notification_handler.base_notification.pipeline_name = input_data['workflow_name']
+        self.notification_handler.base_notification.schedule_name = input_data['schedule_name']
+        self.notification_handler.base_notification.model_name = input_data['model_name']
+        self.notification_handler.base_notification.model_id = input_data['model_id']

laborious/worker/worker.py

@@ -0,0 +1,105 @@
+from temporalio import workflow, client
+from temporalio.worker import Worker
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.workflows.predictions_batch import PredictionsBatch
+    from laborious.activities.activities import Activities
+    import os
+    import logging
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+async def main():
+    host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
+    logger = logging.getLogger(__name__)
+    stream_handler = logging.StreamHandler()
+    stream_handler.setLevel(
+        os.getenv('LOG_LEVEL', 'INFO').upper()
+    )
+    stream_handler.setFormatter(
+        logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        )
+    )
+
+    logger.addHandler(stream_handler)
+
+    notification_handler = NotificationHandler(
+        servers=os.getenv('NOTIFICATION_SERVERS', 'http://localhost:29092'),
+        logger=logger,
+        project_name=os.getenv('PROJECT_NAME', 'laborious'),
+        pipeline_name='-',
+        trigger_name='-',
+        model_name='-',
+        model='-'
+    )
+
+    postgres_config = {
+        'host': os.getenv('POSTGRES_HOST', 'localhost'),
+        'port': int(os.getenv('POSTGRES_PORT', '5432')),
+        'user': os.getenv('POSTGRES_USER', 'postgres'),
+        'password': os.getenv('POSTGRES_PASSWORD', 'postgres'),
+        'dbname': os.getenv('POSTGRES_DBNAME', 'postgres'),
+        'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '5')),
+        'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '20'))
+    }
+
+    mlflow_config = {
+        'host': os.getenv('MLFLOW_HOST', 'localhost'),
+        'port': int(os.getenv('MLFLOW_PORT', '5000')),
+        'username': os.getenv('MLFLOW_USERNAME', 'aignosi'),
+        'password': os.getenv('MLFLOW_PASSWORD', 'aignosi')
+    }
+
+    opc_config = {
+        'name': os.getenv('OPC_NAME', 'opc'),
+        'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'),
+        'server_uri': os.getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
+        'cert_path': os.getenv('OPC_CERT_PATH', None),
+        'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', None),
+        'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', None)
+    }
+
+    activities = Activities(
+        postgres_config=postgres_config,
+        mlflow_config=mlflow_config,
+        opc_config=opc_config,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    temporal_client = await client.Client.connect(target_host=host)
+    workers = [
+        Worker(
+            temporal_client,
+            task_queue='predictions',
+            workflows=[PredictionsBatch],
+            activities=[
+                # Base
+                activities.prepare_activity,
+                # MLFlow
+                activities.request_predict,
+                activities.request_transform,
+                # Gates
+                activities.input_gate,
+                activities.mlflow_response_gate,
+                activities.mlflow_content_gate,
+                activities.format_prediction,
+                activities.format_default_prediction,
+                activities.get_last_timestamp,
+                # OPC
+                activities.write_opc_data,
+                # Postgres
+                activities.load_custom_query,
+                activities.repeat_last_prediction,
+                activities.export_data_to_postgres
+            ]
+        )
+    ]
+
+    for w in workers:
+        await w.run()
+
+if __name__ == '__main__':
+    import asyncio
+    asyncio.run(main())

laborious/workflows/predictions_batch.py

@@ -15,7 +15,8 @@ class PredictionsBatch():
             {
                 'schedule_name': input_data['schedule_name'],
                 'model_name': input_data['model_name'],
-                'model_id': input_data['model_id']
+                'model_id': input_data['model_id'],
+                'workflow_name': 'predictions_batch'
             }
         )
 

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -70,6 +70,7 @@ class FormatAndExportPrediction():
             }
         )
 
+        # write to opc
         opc_holder = workflow.execute_activity_method(
             Activities.write_opc_data,
             {