From 6f81ce7d03805f0b35b8e5e73594461d1e273648 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 27 May 2025 16:49:31 -0300 Subject: [PATCH 1/9] SIENTIAPDE-994 Refactor logging in MLFlow and Gates activities; remove print statement in FormatAndExportPrediction; update data handling in PredictionProcess; delete unused redis-feeder script. --- laborious/activities/gates.py | 2 +- laborious/activities/mlflow.py | 5 ++ .../utils/repository/model_repository.py | 7 ++- .../format_and_export_prediction.py | 2 - .../sub_workflows/prediction_process.py | 13 +++-- simulator/redis-feeder.py | 55 ------------------- 6 files changed, 19 insertions(+), 65 deletions(-) delete mode 100644 simulator/redis-feeder.py diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index c4f9dd5..1cf3fb9 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -73,7 +73,7 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data.to_string()}") + self.logger.debug(f"Input data:\n {data}") self.logger.debug(f"Filters: {filters}") for fil, config in filters.items(): diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index ed96192..ae141a2 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -41,6 +41,7 @@ class MLFlow(BaseActivity): model_name = input_data['model_name'] model_retention = input_data['model_retention'] + self.logger.debug("Raw input data:") self.logger.debug(data) data = data.pivot( @@ -50,9 +51,13 @@ class MLFlow(BaseActivity): data.reset_index(inplace=True) data.columns.name = None + self.logger.debug("Processed input data:") + self.logger.debug(data) + response_data = self.model_monitoring_repository.transform( model_name, data, model_retention) + self.logger.debug("Response data:") self.logger.debug(response_data) return response_data diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 3896411..f2eaeb9 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -13,7 +13,6 @@ import traceback import mlflow import pandas as pd from sientia.ModelServing import ModelServing -from pathlib import Path class MLFlowRepository(): @@ -260,7 +259,8 @@ class MLFlowRepository(): try: return { 'success': True, - 'content': self.model_serving.get_cached_transform(model_name, data, model_retention).to_dict() + 'content': self.model_serving.get_cached_transform( + model_name, data, model_retention).to_dict() } except Exception as e: @@ -276,7 +276,8 @@ class MLFlowRepository(): try: start_time = datetime.now() data = self.model_serving.get_cached_predict( - model_name, data, model_retention) + model_name, data, model_retention)[-1:] + end_time = datetime.now() data = pd.DataFrame(data, columns=['prediction']) data['response_time'] = (end_time - start_time).total_seconds() diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 67cad49..3b1fad7 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -40,8 +40,6 @@ class FormatAndExportPrediction(): data = input_data['data'] prediction_confidence = input_data['prediction_confidence'] - print(f"Input data: {input_data}") - if path_flag is None: # proceed with formatting and exporting prediction = await workflow.execute_local_activity_method( diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index de84328..3164639 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -97,11 +97,13 @@ class PredictionProcess(): ): return + transformed_data = response_data['content'] + path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], - 'data': response_data, + 'data': transformed_data, 'type': 'transform', 'path_priority': input_data['path_priority'] }, @@ -117,7 +119,7 @@ class PredictionProcess(): response_data = await workflow.execute_local_activity_method( Activities.request_predict, { - 'data': response_data, + 'data': transformed_data, 'model_name': model_name, 'model_retention': model_retention }, @@ -148,11 +150,14 @@ class PredictionProcess(): 'path_flag': path_flag, 'data': response_data['content'], 'prediction_confidence': confidence, - 'timestamp': response_data['timestamp'], + 'timestamp': last_timestamp, 'model_id': model_id, 'model_name': model_name, 'model_retention': model_retention, - 'opc_output_config': input_data['opc_output_config'] + 'opc_output_config': input_data['opc_output_config'], + 'schema': input_data['schema'], + 'table_name': input_data['table_name'], + 'comment': comment } ) diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py deleted file mode 100644 index 3f7e36a..0000000 --- a/simulator/redis-feeder.py +++ /dev/null @@ -1,55 +0,0 @@ -import redis -import json -import os - -# Redis connection settings -redis_host = "localhost" -redis_port = 6379 - -# Connect to Redis -r = redis.Redis(host=redis_host, port=redis_port, - decode_responses=True, username='default', password='bdnZOpcyiL') - -# Define the key pattern to target -pattern = "slot:opc_tags:*" - -# Step 1: Find and delete matching keys -print("🔍 Searching for keys matching:", pattern) -for key in r.scan_iter(match=pattern): - r.delete(key) - print(f"❌ Deleted: {key}") - -# Step 2: Insert new data -# Example new OPC tag data -new_data = { - "slot:opc_tags:1": { - "server1": { - "name": "server1", - "url": "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840", - "server_uri": "http://opcua-server.simulator", - "tags": { - 'ns=2;i=2': { - 'tag_name': 'Counter', - 'frequency': 1000, - 'topics': ['opcua', 'counter'], - }, - 'ns=2;i=3': { - 'tag_name': 'Rollout', - 'frequency': 1000, - "topics": ['opcua', 'rollout'], - }, - 'ns=2;i=4': { - 'tag_name': 'Square', - 'frequency': 1000, - "topics": ['opcua'], - }, - } - } - } -} - -for key, val in new_data.items(): - r.set(key, json.dumps(val)) - print(f"✅ Set: {key} -> {val}") - -print("🚀 OPC tag keys replaced successfully.") From 325d22f2419afe2078f47ff1dc3bea7af4fb5ef9 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 May 2025 08:23:50 -0300 Subject: [PATCH 2/9] SIENTIAPDE-994Add shutdown methods for Activities and OPC classes; enhance worker error handling --- laborious/activities/activities.py | 4 + laborious/activities/opc.py | 4 + laborious/worker/worker.py | 15 ++- values.yaml | 183 +++++++++++++++++++++++++++++ 4 files changed, 205 insertions(+), 1 deletion(-) diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index 5ba2cee..e42e847 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -47,3 +47,7 @@ class Activities(Postgres, MLFlow, Gates, OPC): @activity.defn(name="prepare_activity") async def prepare_activity(self, input_data: dict[str, Any]): await super().prepare_activity(input_data) + + def shutdown(self): + Postgres.close(self) + OPC.shutdown(self) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index e6a1dcc..0f02c43 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -99,3 +99,7 @@ class OPC(BaseActivity): data_type=tag_config['data_type'], tag_type='confidence' ) + + def shutdown(self): + for opc in self.opc_repository.values(): + opc.disconnect() diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index bdf168d..4e36cc2 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -1,5 +1,6 @@ from temporalio import workflow, client from temporalio.worker import Worker +import sys with workflow.unsafe.imports_passed_through(): import os @@ -90,7 +91,19 @@ async def main(): logger.info('Workers started successfully') - await asyncio.gather(*handlers) + try: + # This will run the workers and wait for them to complete. + # If an exception occurs in any of the worker handlers, it will be propagated here. + await asyncio.gather(*handlers) + except BaseException as e: + logger.error("An unhandled exception occurred: %s", e, exc_info=True) + finally: + if notification_handler: + notification_handler.shutdown() + if activities: + activities.shutdown() + # Exit with a non-zero status code to indicate failure to Kubernetes + sys.exit(1) if __name__ == '__main__': asyncio.run(main()) diff --git a/values.yaml b/values.yaml index e69de29..7e60b9f 100644 --- a/values.yaml +++ b/values.yaml @@ -0,0 +1,183 @@ +# Default values for sientia-module. +# This is a YAML-formatted file. +# Declare variables to be passed into your templates. + +# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/ +replicaCount: 1 + +# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ +image: + repository: aignosi.azurecr.io/sientia-module + # This sets the pull policy for images. + pullPolicy: Always + # Overrides the image tag whose default is the chart appVersion. + tag: "0.0.2" + +# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ +imagePullSecrets: +- name: docker-hub-secret +# This is to override the chart name. +nameOverride: "sientia-laborious-worker" +fullnameOverride: "sientia-laborious-worker" +namespace: sientia + +# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/ +serviceAccount: + # Specifies whether a service account should be created + create: true + # Automatically mount a ServiceAccount's API credentials? + automount: true + # Annotations to add to the service account + annotations: {} + # The name of the service account to use. + # If not set and create is true, a name is generated using the fullname template + name: "sientia-laborious-worker" + +# This is for setting Kubernetes Annotations to a Pod. +# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/ +podAnnotations: {} +# This is for setting Kubernetes Labels to a Pod. +# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/ +podLabels: {} + +podSecurityContext: {} + # fsGroup: 2000 + +securityContext: {} + # capabilities: + # drop: + # - ALL + # readOnlyRootFilesystem: true + # runAsNonRoot: true + # runAsUser: 1000 + + +resources: {} + # We usually recommend not to specify default resources and to leave this as a conscious + # choice for the user. This also increases chances charts run on environments with little + # resources, such as Minikube. If you do want to specify resources, uncomment the following + # lines, adjust them as necessary, and remove the curly braces after 'resources:'. + # limits: + # cpu: 100m + # memory: 128Mi + # requests: + # cpu: 100m + # memory: 128Mi + +# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/ +livenessProbe: + exec: + command: + - sh + - -c + - pgrep -f "laborious.worker.worker" + initialDelaySeconds: 20 + periodSeconds: 30 + +readinessProbe: + exec: + command: + - sh + - -c + - pgrep -f "laborious.worker.worker" + initialDelaySeconds: 10 + periodSeconds: 15 + + +# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/ +autoscaling: + enabled: false + minReplicas: 1 + maxReplicas: 100 + targetCPUUtilizationPercentage: 80 + # targetMemoryUtilizationPercentage: 80 + +# Additional volumes on the output Deployment definition. +volumes: [] +# - name: foo +# secret: +# secretName: mysecret +# optional: false + +# Additional volumeMounts on the output Deployment definition. +volumeMounts: [] +# - name: foo +# mountPath: "/etc/foo" +# readOnly: true + +nodeSelector: {} + +tolerations: [] + +affinity: {} + +service: + enabled: false + type: ClusterIP + port: 4840 + targetPort: 4840 + + +env: + # Entrypoint variables + - name: GITHUB_REPO_URL + value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" + - name: GITHUB_BRANCH + value: "SIENTIAPDE-994-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas" + - name: PYTHON_APP + value: "laborious.worker.worker" + + # Application variables + - name: POSTGRES_HOST + value: "paradedb-rw.paradedb.svc.cluster.local" + - name: POSTGRES_PORT + value: "5432" + - name: POSTGRES_USER + value: "sientia" + - name: POSTGRES_PASSWORD + value: "sientia" + - name: POSTGRES_DBNAME + value: "sientia" + - name: POSTGRES_MIN_CONNECTIONS + value: "10" + - name: POSTGRES_MAX_CONNECTIONS + value: "20" + + - name: MLFLOW_HOST + value: "sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local" + - name: MLFLOW_PORT + value: "80" + - name: MLFLOW_USERNAME + value: "aignosi" + - name: MLFLOW_PASSWORD + value: "aignosi" + + - name: OPC_NAME + value: "server-1" + - name: OPC_URL + value: "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840" + + - name: LOG_LEVEL + value: "DEBUG" + - name: PROJECT_NAME + value: "sientia-laborious" + + - name: TEMPORAL_HOST + value: "temporal-frontend.temporal.svc.cluster.local:7233" + - name: TEMPORAL_NAMESPACE + value: "default" + +ssh: + enabled: true + secretName: git-ssh-key-sientia-laborious-worker + sshPath: /mnt/.ssh + knownHostsPath: /mnt/known_hosts + +# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp + +# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat + +# kubectl create secret generic git-ssh-key-sientia-laborious-worker \ +# --namespace sientia \ +# --from-file=ssh-privatekey=git_key \ +# --type=kubernetes.io/ssh-auth \ No newline at end of file From b3db8d119031055a19ab2ac782bb33c7532f721a Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 May 2025 08:54:16 -0300 Subject: [PATCH 3/9] ignore keys --- .gitignore | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index a446ab3..f9bab0b 100644 --- a/.gitignore +++ b/.gitignore @@ -36,4 +36,7 @@ __pycache__/ # Ignorar coverage htmlcov/ -.coverage \ No newline at end of file +.coverage + +# git keys +git_key* \ No newline at end of file From cfb217963b5a7d40327028f7cb91c5317254f2b3 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 May 2025 09:24:16 -0300 Subject: [PATCH 4/9] SIENTIAPDE-994 Add shutdown tests for Activities and OPC classes; improve test assertions --- tests/laborious/activities/test_activities.py | 44 +++++++++++++++++++ tests/laborious/activities/test_opc.py | 5 +++ .../utils/repository/test_model_repository.py | 4 +- .../subworkflows/test_prediction_process.py | 15 ++++--- 4 files changed, 60 insertions(+), 8 deletions(-) diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index 98be011..3b2ef49 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -147,3 +147,47 @@ async def test_prepare_activity(_mock_opc_init, 'model_name'] assert activities.notification_handler.base_notification.model_id == input_data[ 'model_id'] + + +@patch('laborious.activities.activities.Postgres', return_value=MagicMock()) +@patch('laborious.activities.activities.MLFlow', return_value=MagicMock()) +@patch('laborious.activities.activities.OPC', return_value=MagicMock()) +def test_shutdown(mock_opc_init, + _mock_mlflow_init, mock_postgres_init): + postgres_config = { + 'host': 'localhost', + 'port': 5432, + 'user': 'postgres', + 'password': 'postgres', + 'dbname': 'postgres', + 'min_connections': 1, + 'max_connections': 10 + } + + mlflow_config = { + 'host': 'localhost', + 'port': 5000, + 'username': 'mlflow', + 'password': 'mlflow' + } + + opc_config = { + 'bootstrap_servers': 'localhost:9092', + 'polling_time': 1000, + 'group_id': 'test-group' + } + + logger = MagicMock() + notification_handler = MagicMock() + + activities = Activities( + postgres_config=postgres_config, + mlflow_config=mlflow_config, + opc_config=opc_config, + logger=logger, + notification_handler=notification_handler + ) + + activities.shutdown() + mock_opc_init.shutdown.assert_called_once() + mock_postgres_init.close.assert_called_once() diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index c4e26ee..d012778 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -189,3 +189,8 @@ async def test_write_opc_data_empty_config(opc): # Assert opc.opc_repository['server1'].write_data.assert_not_called() + + +def test_shutdown(opc): + opc.shutdown() + opc.opc_repository['server1'].disconnect.assert_called_once() diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index abf4ab8..a675edb 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -251,9 +251,9 @@ def test_predict_success(mlflow_repository): mlflow_repository.model_serving.get_cached_predict.assert_called_once_with( model_name, data, 1) - assert output['success'] == True + assert output['success'] is True assert output['content'] == {'prediction': { - 0: 2, 1: 3}, 'response_time': ANY} + 0: 3}, 'response_time': ANY} def test_predict_error(mlflow_repository): diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 4aebc6f..4318379 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -73,13 +73,13 @@ async def test_run(workflow_mock, prediction_process): workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], - 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, + 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'] }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { - 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, + 'data': 'transformed_data', 'model_name': input_data['model_name'], 'model_retention': input_data['model_retention'] }, retry_policy=ANY, start_to_close_timeout=ANY)]) @@ -101,7 +101,10 @@ async def test_run(workflow_mock, prediction_process): 'model_id': 1, 'model_name': 'test_model_name', 'model_retention': '30', - 'opc_output_config': input_data['opc_output_config'] + 'opc_output_config': input_data['opc_output_config'], + 'schema': input_data['schema'], + 'table_name': input_data['table_name'], + 'comment': 'Error' } ) @@ -268,7 +271,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], - 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, + 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'] }, retry_policy=ANY, start_to_close_timeout=ANY)]) @@ -338,13 +341,13 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], - 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, + 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'] }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { - 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, + 'data': 'transformed_data', 'model_name': input_data['model_name'], 'model_retention': input_data['model_retention'] }, retry_policy=ANY, start_to_close_timeout=ANY)]) From 36a2fbbc13b1798074f1a662bd4f5d62f97b3431 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 May 2025 09:40:05 -0300 Subject: [PATCH 5/9] name and variable fixes --- laborious/worker/worker.py | 2 +- values.yaml | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 4e36cc2..5b902a9 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -28,7 +28,7 @@ async def main(): logger.info('Starting Notification Handler...') notification_handler = NotificationHandler( - servers=os.getenv('KAFKA_SERVERS', 'http://localhost:9092'), + servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), logger=logger, project_name=os.getenv('PROJECT_NAME', 'laborious'), pipeline_name='-', diff --git a/values.yaml b/values.yaml index 7e60b9f..99c82f5 100644 --- a/values.yaml +++ b/values.yaml @@ -157,6 +157,9 @@ env: - name: OPC_URL value: "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840" + - name: KAFKA_BOOTSTRAP_SERVERS + value: "kafka.kafka.svc.cluster.local:9092" + - name: LOG_LEVEL value: "DEBUG" - name: PROJECT_NAME From 11f41f358a4fc0fac8a26eada8581eeb3350a155 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 May 2025 09:59:22 -0300 Subject: [PATCH 6/9] values fix --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 99c82f5..b164156 100644 --- a/values.yaml +++ b/values.yaml @@ -144,7 +144,7 @@ env: value: "20" - name: MLFLOW_HOST - value: "sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local" + value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local" - name: MLFLOW_PORT value: "80" - name: MLFLOW_USERNAME From 5326051714fc6cc0671720cfb8c2e2fc34b72190 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 9 Jun 2025 11:08:07 -0300 Subject: [PATCH 7/9] SIENTIAPDE-1094 Refactor activity imports and remove unused base and logger files; update requirements for library versioning --- laborious/activities/activities.py | 4 +- laborious/activities/base.py | 26 -- laborious/activities/gates.py | 2 +- laborious/activities/mlflow.py | 4 +- laborious/activities/opc.py | 2 +- laborious/activities/postgres.py | 181 ------------- laborious/utils/logger.py | 22 -- laborious/utils/policies.py | 9 - .../utils/repository/model_repository.py | 241 +----------------- laborious/worker/worker.py | 5 +- laborious/workflows/predictions_batch.py | 2 +- .../format_and_export_prediction.py | 2 +- .../sub_workflows/prediction_process.py | 2 +- requirements.txt | 2 +- tests/laborious/activities/test_activities.py | 6 +- tests/laborious/activities/test_base.py | 35 --- tests/laborious/activities/test_postgres.py | 159 ------------ .../utils/repository/test_model_repository.py | 190 +------------- tests/laborious/utils/test_logger.py | 37 --- 19 files changed, 23 insertions(+), 908 deletions(-) delete mode 100644 laborious/activities/base.py delete mode 100644 laborious/activities/postgres.py delete mode 100644 laborious/utils/logger.py delete mode 100644 laborious/utils/policies.py delete mode 100644 tests/laborious/activities/test_base.py delete mode 100644 tests/laborious/activities/test_postgres.py delete mode 100644 tests/laborious/utils/test_logger.py diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index e42e847..d3c9436 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -1,13 +1,13 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from laborious.activities.postgres import Postgres + from sientia_do.temporal.activities.postgres import Postgres + from sientia_do.notifications.handlers import NotificationHandler from laborious.activities.mlflow import MLFlow from laborious.activities.gates import Gates from laborious.activities.opc import OPC from typing import Any from logging import Logger - from sientia_do.notifications.handlers import NotificationHandler class Activities(Postgres, MLFlow, Gates, OPC): diff --git a/laborious/activities/base.py b/laborious/activities/base.py deleted file mode 100644 index 3adb0e4..0000000 --- a/laborious/activities/base.py +++ /dev/null @@ -1,26 +0,0 @@ -from typing import Any -from logging import Logger -from temporalio import activity -from sientia_do.notifications.handlers import NotificationHandler - - -class BaseActivity: - def __init__(self, logger: Logger, notification_handler: NotificationHandler): - self.logger = logger - self.notification_handler = notification_handler - - @activity.defn(name="prepare_activity") - async 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.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'] diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 1cf3fb9..799fed5 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -6,7 +6,7 @@ with workflow.unsafe.imports_passed_through(): from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel - from laborious.activities.base import BaseActivity + from sientia_do.temporal.activities.base import BaseActivity from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter from typing import Any from laborious.utils.filters.conditional_filters import ( diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index ae141a2..a209ed1 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -4,11 +4,11 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from laborious.activities.base import BaseActivity + from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.notifications.handlers import NotificationHandler from laborious.utils.repository.model_repository import MLFlowRepository from typing import Any from logging import Logger - from sientia_do.notifications.handlers import NotificationHandler class MLFlow(BaseActivity): diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 0f02c43..997b7c2 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through(): from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel - from laborious.activities.base import BaseActivity + from sientia_do.temporal.activities.base import BaseActivity from laborious.utils.repository.opc_repository import OpcRepository from typing import Any import traceback diff --git a/laborious/activities/postgres.py b/laborious/activities/postgres.py deleted file mode 100644 index 9fc0ff7..0000000 --- a/laborious/activities/postgres.py +++ /dev/null @@ -1,181 +0,0 @@ -import traceback -from temporalio import workflow, activity - -from laborious.activities.base import BaseActivity -with workflow.unsafe.imports_passed_through(): - from sqlalchemy import create_engine - from sqlalchemy.orm import sessionmaker - from sqlalchemy.pool import QueuePool - from psycopg2.pool import ThreadedConnectionPool - from pandas import read_sql_query, DataFrame - from logging import Logger - from sientia_do.notifications.handlers import NotificationHandler - from sientia_do.notifications.models import NotificationLevel - from typing import Any - - -class Postgres(BaseActivity): - def __init__(self, host: str, port: int, - user: str, password: str, dbname: str, - min_connections: int, max_connections: int, - logger: Logger, notification_handler: NotificationHandler): - self.host = host - self.port = port - self.user = user - self.password = password - self.dbname = dbname - - # Create SQLAlchemy engine with connection pooling - self.engine = create_engine( - f'postgresql://{user}:{password}@{host}:{port}/{dbname}', - poolclass=QueuePool, - pool_size=min_connections, - max_overflow=max_connections - min_connections, - pool_pre_ping=True - ) - self.session_factory = sessionmaker(bind=self.engine) - - BaseActivity.__init__(self, logger, notification_handler) - - def close(self): - self.engine.dispose() - - def __del__(self): - self.close() - - @activity.defn(name="load_custom_query") - async def load_custom_query(self, query: str) -> dict[str, Any]: - """ - Loads data from a custom query. - - Args: - query (str): The query to load data from. - - Returns: - dict[str, dict]: The data from the query. - """ - self.logger.info(f"Fetching data from query: {query}") - - data = None - with self.session_factory() as session: - try: - data = read_sql_query(query, self.engine) - - except Exception as e: - trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( - notification_id="ERROR_LOADING_CUSTOM_QUERY", - message=f"Error fetching data from query: {e}", - block="load_custom_query", - level=NotificationLevel.ERROR, - attachment_content=trace - ) - - self.logger.error(trace) - - return {} - finally: - session.close() - - if data is None: - return {} - - # Converts any datetime datatype columns to string - for col in data.select_dtypes(include=['datetime64']).columns: - data[col] = data[col].dt.strftime('%Y-%m-%d %H:%M:%S') - - self.logger.info(f"Fetched {len(data)} rows") - self.logger.debug(f"Data: \n{data.to_string()}") - - return data.to_dict() - - @activity.defn(name="repeat_last_prediction") - async def repeat_last_prediction(self, query_items: dict[str, str]): - """ - Repeats the last prediction for a given model. - - Args: - query_items (dict[str, str]): The query items. Contains: - schema (str): The schema of the table. - table_name (str): The name of the table. - model (int): The model to repeat the prediction for. - - Returns: - None - """ - schema = query_items["schema"] - table_name = query_items["table_name"] - model = query_items["model"] - - repeat_query = f""" - INSERT INTO \"{schema}\".{table_name} (model_id, prediction, timestamp, response_time, prediction_status, prediction_confidence, created_at) - SELECT model_id, prediction, timestamp, response_time, prediction_status, prediction_confidence, NOW() - FROM \"{schema}\".{table_name} - WHERE model_id = {model} - ORDER BY timestamp DESC - LIMIT 1; - """ - self.logger.info(f"Repeating last prediction for model {model}") - self.logger.debug(f"Query: {repeat_query}") - - with self.session_factory() as session: - try: - session.execute(repeat_query) - session.commit() - - except Exception as e: - trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( - notification_id="ERROR_REPEATING_LAST_PREDICTION", - message=f"Error repeating last prediction: {e}", - block="repeat_last_prediction", - level=NotificationLevel.ERROR, - attachment_content=trace - ) - - self.logger.error(trace) - - finally: - session.close() - - @activity.defn(name="export_data_to_postgres") - async def export_data_to_postgres(self, input_data: dict[str, Any]): - """ - Exports data to a postgres table. - - Args: - input_data (dict[str, Any]): The data to export. Contains: - schema (str): The schema of the table. - table_name (str): The name of the table. - data (DataFrame): The data to export. - """ - - self.logger.debug( - f"Exporting data to postgres: {input_data['data']}") - - schema = input_data["schema"] - table_name = input_data["table_name"] - data = DataFrame(input_data["data"]) - - with self.session_factory() as session: - try: - data.to_sql(table_name, self.engine, schema=schema, - if_exists="append", index=False) - session.commit() - - except Exception as e: - trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( - notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES", - message=f"Error exporting data to postgres: {e}", - block="export_data_to_postgres", - level=NotificationLevel.ERROR, - attachment_content=trace - ) - - self.logger.error(trace) - - else: - self.logger.debug("Data exported to postgres") - finally: - session.close() diff --git a/laborious/utils/logger.py b/laborious/utils/logger.py deleted file mode 100644 index 42a9cfd..0000000 --- a/laborious/utils/logger.py +++ /dev/null @@ -1,22 +0,0 @@ -from os import getenv -import logging -import sys - - -def get_logger(name: str): - log_level = getenv('LOG_LEVEL', 'INFO').upper() - - logger = logging.getLogger(name) - logger.setLevel(log_level) - stream_handler = logging.StreamHandler(sys.stdout) - stream_handler.setLevel(log_level) - - stream_handler.setFormatter( - logging.Formatter( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - ) - ) - - logger.addHandler(stream_handler) - - return logger diff --git a/laborious/utils/policies.py b/laborious/utils/policies.py deleted file mode 100644 index 8c7449a..0000000 --- a/laborious/utils/policies.py +++ /dev/null @@ -1,9 +0,0 @@ -from datetime import timedelta -from temporalio.common import RetryPolicy - -retry_policy = RetryPolicy( - initial_interval=timedelta(seconds=1), - backoff_coefficient=2.0, - maximum_interval=timedelta(minutes=1), - maximum_attempts=1 -) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index f2eaeb9..ed5936b 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -1,16 +1,17 @@ """ Model Monitoring Repository -This module contains the ModelMonitoringRepository class, which is responsible for handling the communication with the Model Monitoring API. +This module contains the ModelMonitoringRepository class, which is responsible +for handling the communication with the Model Monitoring API. -It includes the methods that are used to answer ModelMonitoringService requests using the Model Monitoring API functions. +It includes the methods that are used to answer ModelMonitoringService requests using +the Model Monitoring API functions. By Monitoring we mean the evaluation of the performance of models, the generation of reports. """ from datetime import datetime import traceback -import mlflow import pandas as pd from sientia.ModelServing import ModelServing @@ -21,240 +22,6 @@ class MLFlowRepository(): self.model_serving = ModelServing(tracking_uri=host, username=username, password=password) - def get_current_data_df(self, current_data: pd.DataFrame, model_name: str, target: str): - """ - Get the current data as a DataFrame and update the prediction and target columns - - Parameters: - current_data (pd.DataFrame): the current data - model_name (str): the name of the model - target (str): the target column - - Returns: - DataFrame: the current data as a DataFrame - - - """ - predictions = current_data['prediction'] - - target = current_data[target] - current_data = self.model_serving.get_transformed_data( - model_name, current_data, by='model') - current_data['prediction'] = predictions - current_data['target'] = target - - return pd.DataFrame(current_data).dropna() - - def get_artifact(self, destination: str, search_by: str, run_id: str = None, - model_name: str = None, artifact_name: str = None) -> None: - """ - Get an artifact in MLflow by experiment or model and save it to a destination path using API. - If the artifact is searched by model, the latest production version will be used. - - Args: - destination: The destination path to save the artifact. - search_by: The way to search for the artifact ('experiment' or 'model'). - run_id: The run ID of the experiment (if search_by is "experiment"). - model_name: The name of the model (if search_by is "model"). - artifact_name: The path of the artifact to download. - - Returns: - artifact: The artifact(.csv) downloaded from MLflow. - """ - - self.model_serving.get_artifact(destination=destination, search_by=search_by, - run_id=run_id, model_name=model_name, artifact_name=artifact_name) - - def calculate_model_metrics(self, real_data, predictions, flag): - """ - Function to calculate the metrics of a model using API - - Parameters: - real_data (array): the real data - predictions (array): the predictions - - Returns: - dict: the metrics of the model including MSE and R2 - """ - return self.model_serving.get_model_metrics(reference_data=None, real_data=real_data, predictions=predictions, type_flag=flag) - - def get_experiment_by_run_id(self, run_id: str) -> dict: - # Get the run information using the run_id - run = mlflow.get_run(run_id) - - # Extract the experiment ID from the run - experiment_id = run.info.experiment_id - - # Get the experiment details using the experiment ID - experiment = mlflow.get_experiment(experiment_id) - experiment_name = experiment.name - return experiment_name - - def get_next_run_name(self, model_name: str) -> str: - """ - Function to get the next run number of a specific model - - Parameters: - model_name (str): the name of the model - - Returns: - str: the next run number - """ - - runs = mlflow.search_runs( - experiment_names=[model_name], order_by=["start_time desc"]) - next_run_number = len(runs) + 1 - return f"{model_name}-{next_run_number}" - - def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple: - """ - Retrain a model with new data. - - Parameters: - data (pandas.DataFrame): The new data to use for retraining. - model_name (str): The name of the model to retrain. - metrics_list (list): The metrics to be used to compare the models. - compare_metrics (bool): If True, the retrain will only be considered if the new model is better than the current one. - If False, the retrain will always be considered. - split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets. - If False, the data will be used as a unique block for retraining. - update_report (bool): If True, a report will be created with the data of the retrained model. - update_transformation (bool): If True, the model will be updated in the MLflow tracking server. - update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server. - shuffle_data (bool): If True, the data will be shuffled before splitting. - model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'. - - - Returns: - mlflow.sklearn.Model: The retrained prediction model. - mlflow.sklearn.Model: The retrained data model. - mse (float): The mean squared error of the retrained model. - r2 (float): The R-squared score of the retrained model. - """ - - # load predictor model - predictor_uri = f"models:/{model_name}/production" - # load transform model - latest_production_id = self.model_serving.get_model_run_id( - model_name, stage="Production" - ) - transform_uri = self.model_serving.get_model_uri( - latest_production_id, prediction=False - ) - # load - data_model = mlflow.sklearn.load_model(transform_uri) - prediction_model = mlflow.sklearn.load_model(predictor_uri) - data_model = data_model.fit(data) - treated_data = data_model.predict(data) - # align target column with treated_data - target_name = data_model.target_variable - y = data[target_name] - treated_data = pd.merge( - treated_data, y, left_index=True, right_index=True) - prediction_model = prediction_model.fit(treated_data) - # Example usage - experiment = self.get_experiment_by_run_id(latest_production_id) - pred_model_atributes = vars(prediction_model) # load class attributes - data_model_atributes = vars(data_model) # load class attributes - mlflow.set_experiment(experiment) - experiment_description = "Retrain model {model_name} with new data" - current_run_name = self.get_next_run_name(experiment) - with mlflow.start_run( - run_name=current_run_name, description=experiment_description - ) as _run: - # update transfomation model - # fixed parameters - for name_atribute, val_atribute in pred_model_atributes.items(): - if name_atribute != "model": - mlflow.log_param(name_atribute, val_atribute) - # update prediction model - for name_atribute, val_atribute in data_model_atributes.items(): - if name_atribute != "model": - mlflow.log_param(name_atribute, val_atribute) - # dynamic parameters, including model itself - mlflow.sklearn.log_model(data_model, "data_model") - file_path = f"laborious/data/raw_data_{model_name}.csv" - data.to_csv( - f"laborious/data/raw_data_{model_name}.csv", index=True) - # log the data raw - mlflow.log_artifact(file_path) - - # dynamic parameters, including model itself - mlflow.sklearn.log_model(prediction_model, "prediction_model") - mlflow.log_param("retrain", True) - - return "Model retrained successfully", experiment - - def get_experiment(self, experiment_name: str) -> int: - experiment = mlflow.get_experiment_by_name(experiment_name) - - if experiment is None: - raise ValueError(f'Experiment {experiment_name} not found') - - return int(experiment.experiment_id) - - def get_experiment_last_run(self, experiment_id: int) -> str: - runs = mlflow.search_runs( - experiment_ids=[experiment_id], - filter_string="", # Sem filtro no MLflow ainda - output_format="pandas" - ) - - # Filtrar apenas as runs onde params.retrain == True - filtered_runs = runs[runs["params.retrain"] == 'True'] - - # Converter a coluna 'end_time' para datetime - filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time']) - - # Ordenar o DataFrame de forma descendente pela coluna 'end_time' - filtered_runs = filtered_runs.sort_values( - by='end_time', ascending=False) - - # Pegar a última run_id do DataFrame filtrado e ordenado - latest_run_id = filtered_runs.iloc[0]['run_id'] - - return latest_run_id - - def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict: - # Registrar o modelo - # Aqui estamos assumindo que você já tem um modelo salvo, caso contrário você precisará treiná-lo e salvá-lo primeiro. - # Se o modelo já está registrado, você pode usar o método register_model() ou pyfunc.load_model() para isso. - mlflow.register_model( - f"runs:/{run_id}/prediction_model", model_name) - - # Colocar a versão do modelo em produção - # Depois de registrar o modelo, precisamos pegar a versão mais recente do modelo e movê-lo para o estágio 'Production' - client = mlflow.tracking.MlflowClient() - - # Obter a versão mais recente registrada do modelo - model_versions = client.get_registered_model( - model_name).latest_versions - max_version = max(model_versions, key=lambda x: int(x.version)).version - - # Mover a versão mais recente do modelo para o estágio de 'Production' - client.transition_model_version_stage( - name=model_name, - version=max_version, - stage="Production", - archive_existing_versions=True - ) - - return { - 'model_name': model_name, - 'version': max_version, - 'mlflow_run_id': run_id - } - - def update_production_model(self, experiment: str, model_name: str) -> dict: - - experiment_id = self.get_experiment(experiment) - run_id = self.get_experiment_last_run(experiment_id) - metadata = self.update_production_model_by_run_id(run_id, model_name) - - metadata['mlflow_experiment_id'] = experiment_id - - return metadata - def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): try: return { diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 5b902a9..f10716e 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -1,22 +1,23 @@ from temporalio import workflow, client from temporalio.worker import Worker -import sys + with workflow.unsafe.imports_passed_through(): import os + import sys import asyncio from laborious.workflows.predictions_batch import PredictionsBatch from laborious.workflows.sub_workflows.prediction_process import PredictionProcess from laborious.workflows.sub_workflows.format_and_export_prediction import \ FormatAndExportPrediction from laborious.activities.activities import Activities - from laborious.utils.logger import get_logger from laborious.utils.connectors_config import ( build_postgres_config, build_mlflow_config, build_opc_config ) from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import get_logger async def main(): diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 425f59f..9c4ab61 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -3,7 +3,7 @@ from temporalio import workflow with workflow.unsafe.imports_passed_through(): from laborious.activities.activities import Activities from typing import Any - from laborious.utils.policies import retry_policy + from sientia_do.temporal.utils.policies import retry_policy from datetime import timedelta diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 3b1fad7..68a5958 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -4,7 +4,7 @@ with workflow.unsafe.imports_passed_through(): from laborious.activities.activities import Activities from typing import Any from datetime import timedelta - from laborious.utils.policies import retry_policy + from sientia_do.temporal.utils.policies import retry_policy @workflow.defn(name="format_and_export_prediction") diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 3164639..df740b3 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -3,7 +3,7 @@ from temporalio import workflow with workflow.unsafe.imports_passed_through(): from laborious.activities.activities import Activities from typing import Any - from laborious.utils.policies import retry_policy + from sientia_do.temporal.utils.policies import retry_policy from datetime import timedelta diff --git a/requirements.txt b/requirements.txt index 90e049b..5604fc7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,5 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index 3b2ef49..31f83e8 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -1,7 +1,7 @@ from pytest import mark from unittest.mock import patch, MagicMock, ANY +from sientia_do.temporal.activities.postgres import Postgres from laborious.activities.activities import Activities -from laborious.activities.postgres import Postgres from laborious.activities.mlflow import MLFlow from laborious.activities.gates import Gates from laborious.activities.opc import OPC @@ -139,9 +139,9 @@ async def test_prepare_activity(_mock_opc_init, await activities.prepare_activity(input_data) - assert activities.notification_handler.base_notification.pipeline_name == input_data[ + assert activities.notification_handler.base_notification.pipeline == input_data[ 'workflow_name'] - assert activities.notification_handler.base_notification.schedule_name == input_data[ + assert activities.notification_handler.base_notification.trigger == input_data[ 'schedule_name'] assert activities.notification_handler.base_notification.model_name == input_data[ 'model_name'] diff --git a/tests/laborious/activities/test_base.py b/tests/laborious/activities/test_base.py deleted file mode 100644 index 6978acb..0000000 --- a/tests/laborious/activities/test_base.py +++ /dev/null @@ -1,35 +0,0 @@ -from unittest.mock import MagicMock -from laborious.activities.base import BaseActivity -from pytest import fixture, mark -from sientia_do.notifications.models import Notification - - -@fixture -def base_activity(): - return BaseActivity( - logger=MagicMock(), - notification_handler=MagicMock(), - ) - - -@mark.asyncio -async def test_prepare_activity(base_activity): - base_activity.notification_handler.base_notification = Notification( - project="project", - pipeline="pipeline", - trigger="-", - model_name="-", - model_id="-", - ) - - await base_activity.prepare_activity({ - 'workflow_name': 'test_workflow', - 'schedule_name': 'test_schedule', - 'model_name': 'test_model', - 'model_id': 'test_model_id' - }) - - assert base_activity.notification_handler.base_notification.schedule_name == "test_schedule" - assert base_activity.notification_handler.base_notification.model_name == "test_model" - assert base_activity.notification_handler.base_notification.model_id == "test_model_id" - assert base_activity.notification_handler.base_notification.pipeline_name == "test_workflow" diff --git a/tests/laborious/activities/test_postgres.py b/tests/laborious/activities/test_postgres.py deleted file mode 100644 index e4a4545..0000000 --- a/tests/laborious/activities/test_postgres.py +++ /dev/null @@ -1,159 +0,0 @@ -from unittest.mock import MagicMock, patch -from pytest import fixture, mark -import pandas as pd -from laborious.activities.postgres import Postgres - - -@fixture -@patch("laborious.activities.postgres.create_engine") -def postgres_activity(_mock_create_engine): - return Postgres( - host="localhost", - port=5432, - user="test_user", - password="test_password", - dbname="test_db", - min_connections=1, - max_connections=5, - logger=MagicMock(), - notification_handler=MagicMock() - ) - - -@mark.asyncio -@patch("laborious.activities.postgres.read_sql_query") -async def test_load_custom_query_none_data(mock_read_sql_query, postgres_activity): - query = "SELECT * FROM test_table LIMIT 1" - mock_read_sql_query.return_value = None - - result = await postgres_activity.load_custom_query(query) - - assert isinstance(result, dict) - assert len(result) == 0 - - -@mark.asyncio -@patch("laborious.activities.postgres.read_sql_query") -async def test_load_custom_query_date_converted(mock_read_sql_query, postgres_activity): - query = "SELECT * FROM test_table LIMIT 1" - mock_data = pd.DataFrame({"column1": [1], "column2": ["test"]}) - mock_data['date'] = pd.to_datetime('2022-01-01') - - mock_read_sql_query.return_value = mock_data - - result = await postgres_activity.load_custom_query(query) - - assert isinstance(result, dict) - assert len(result) == 3 - assert "column1" in result - assert "column2" in result - assert "date" in result - assert result['date'] == {0: '2022-01-01 00:00:00'} - - -@mark.asyncio -@patch("laborious.activities.postgres.read_sql_query") -async def test_load_custom_query_success(mock_read_sql_query, postgres_activity): - query = "SELECT * FROM test_table LIMIT 1" - mock_data = pd.DataFrame({"column1": [1], "column2": ["test"]}) - - mock_read_sql_query.return_value = mock_data - - result = await postgres_activity.load_custom_query(query) - - assert isinstance(result, dict) - assert len(result) == 2 - assert "column1" in result - assert "column2" in result - postgres_activity.logger.info.assert_called() - - -@mark.asyncio -async def test_load_custom_query_error(postgres_activity): - query = "SELECT * FROM non_existent_table" - error_msg = "Table not found" - - with patch("laborious.activities.postgres.read_sql_query", side_effect=ValueError(error_msg)): - result = await postgres_activity.load_custom_query(query) - - assert isinstance(result, dict) - assert len(result) == 0 - postgres_activity.notification_handler.build_and_send_notification.assert_called_once() - postgres_activity.logger.error.assert_called() - - -@mark.asyncio -async def test_repeat_last_prediction_success(postgres_activity): - query_items = { - "schema": "public", - "table_name": "predictions", - "model": 1 - } - - with patch("sqlalchemy.orm.session.Session.execute") as mock_execute: - await postgres_activity.repeat_last_prediction(query_items) - - mock_execute.assert_called_once() - postgres_activity.logger.info.assert_called() - - -@mark.asyncio -async def test_repeat_last_prediction_error(postgres_activity): - query_items = { - "schema": "public", - "table_name": "predictions", - "model": 1 - } - error_msg = "Database error" - - with patch("sqlalchemy.orm.session.Session.execute", side_effect=ValueError(error_msg)): - await postgres_activity.repeat_last_prediction(query_items) - - postgres_activity.notification_handler.build_and_send_notification.assert_called_once() - postgres_activity.logger.error.assert_called() - - -@mark.asyncio -async def test_export_data_to_postgres_success(postgres_activity): - input_data = { - "schema": "public", - "table_name": "test_table", - "data": pd.DataFrame({"column1": [1, 2], "column2": ["a", "b"]}) - } - - with patch("laborious.activities.postgres.DataFrame.to_sql") as mock_to_sql: - await postgres_activity.export_data_to_postgres(input_data) - - mock_to_sql.assert_called_once() - postgres_activity.logger.debug.assert_called() - - -@mark.asyncio -async def test_export_data_to_postgres_error(postgres_activity): - input_data = { - "schema": "public", - "table_name": "test_table", - "data": pd.DataFrame({"column1": [1, 2], "column2": ["a", "b"]}) - } - error_msg = "Export failed" - - with patch("laborious.activities.postgres.DataFrame.to_sql", side_effect=ValueError(error_msg)): - await postgres_activity.export_data_to_postgres(input_data) - - postgres_activity.notification_handler.build_and_send_notification.assert_called_once() - postgres_activity.logger.error.assert_called() - - -@mark.asyncio -async def test_close(postgres_activity): - postgres_activity.close() - - postgres_activity.engine.dispose.assert_called_once() - - -@mark.asyncio -async def test_del(postgres_activity): - postgres_activity.close = MagicMock() - postgres_activity.__del__() - - postgres_activity.close.assert_called_once() diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index a675edb..6d10c1c 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -1,14 +1,14 @@ from unittest.mock import ANY, MagicMock, patch import numpy as np -from pandas import DataFrame import pytest from laborious.utils.repository.model_repository import MLFlowRepository @pytest.fixture def mlflow_repository(): - with patch('laborious.utils.repository.model_repository.ModelServing', autospec=True) as MockModelServing: - mock_instance = MockModelServing.return_value + with patch('laborious.utils.repository.model_repository.ModelServing', + autospec=True) as mock_model_serving: + mock_instance = mock_model_serving.return_value mock_instance.get_transformed_data = MagicMock() repo = MLFlowRepository( @@ -19,190 +19,6 @@ def mlflow_repository(): return repo -def test_get_current_data_df(mlflow_repository): - current_data = { - 'prediction': [1, 3], - 'target': [1, 1], - } - mlflow_repository.model_serving.get_transformed_data.return_value = { - 'var1': [1, 2], - 'var2': [2, np.nan], - } - expected = DataFrame({ - 'var1': [1], - 'var2': [2], - 'prediction': [1], - 'target': [1], - }) - output = mlflow_repository.get_current_data_df(current_data, - 'model', 'target') - - mlflow_repository.model_serving.get_transformed_data.assert_called_once_with( - 'model', current_data, by='model') - - diff = output.compare(expected) - assert diff.empty - - -def test_get_artifact(mlflow_repository): - mlflow_repository.get_artifact( - 'destination', 'search_by', 'run_id', 'model', 'artifact' - ) - mlflow_repository.model_serving.get_artifact.assert_called_once_with( - destination='destination', - search_by='search_by', - run_id='run_id', - model_name='model', - artifact_name='artifact' - ) - - -def test_calculate_model_metrics(mlflow_repository): - mlflow_repository.model_serving.get_model_metrics.return_value = 'data' - real_data = 'real_data' - predictions = 'predictions' - flag = 'flag' - output = mlflow_repository.calculate_model_metrics( - real_data, predictions, flag - ) - mlflow_repository.model_serving.get_model_metrics.assert_called_once_with( - reference_data=None, - real_data=real_data, - predictions=predictions, - type_flag=flag - ) - assert output == 'data' - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_get_experiment_by_run_id(mlflow, mlflow_repository): - mlflow.get_run.return_value = MagicMock( - info=MagicMock( - experiment_id='0', - ) - ) - mlflow.get_experiment.return_value = MagicMock() - mlflow.get_experiment.return_value.name = 'test' - - output = mlflow_repository.get_experiment_by_run_id('0') - assert output == 'test' - mlflow.get_run.assert_called_once_with('0') - mlflow.get_experiment.assert_called_once_with('0') - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_get_next_run_name(mlflow, mlflow_repository): - mlflow.search_runs.return_value = [1, 2, 3] - output = mlflow_repository.get_next_run_name('run') - assert output == 'run-4' - mlflow.search_runs.assert_called_once_with( - experiment_names=['run'], - order_by=['start_time desc'], - ) - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_get_experiment_success(mlflow, mlflow_repository): - mlflow.get_experiment_by_name.return_value = MagicMock( - experiment_id='0') - - output = mlflow_repository.get_experiment('test') - - assert output == 0 - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_get_experiment_error(mlflow, mlflow_repository): - mlflow.get_experiment_by_name.return_value = None - - try: - mlflow_repository.get_experiment('test') - except ValueError as e: - assert str(e) == 'Experiment test not found' - else: - assert False - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_get_experiment_last_run(mlflow, mlflow_repository): - mlflow.search_runs.return_value = DataFrame({ - 'params.retrain': ['True', 'False', 'True', 'False'], - 'end_time': ['2021-01-01', '2021-01-02', '2021-01-03', '2021-01-04'], - 'run_id': ['0', '1', '2', '3'], - }) - - output = mlflow_repository.get_experiment_last_run(0) - - mlflow.search_runs.assert_called_once_with( - experiment_ids=[0], - filter_string="", - output_format="pandas", - ) - - assert output == '2' - - -@patch('laborious.utils.repository.model_repository.mlflow') -def test_update_production_model_by_run_id(mlflow, mlflow_repository): - client_mock = MagicMock() - mlflow.tracking.MlflowClient.return_value = client_mock - - client_mock.get_registered_model.return_value = MagicMock( - latest_versions=[ - MagicMock(version='1'), - MagicMock(version='2'), - MagicMock(version='3'), - ] - ) - output = mlflow_repository.update_production_model_by_run_id('0', 'test') - - mlflow.register_model.assert_called_once_with( - "runs:/0/prediction_model", - 'test', - ) - - mlflow.tracking.MlflowClient.assert_called_once() - client_mock.get_registered_model.assert_called_once_with('test') - client_mock.transition_model_version_stage.assert_called_once_with( - name='test', - version='3', - stage='Production', - archive_existing_versions=True, - ) - - assert output == { - 'model_name': 'test', - 'version': '3', - 'mlflow_run_id': '0', - } - - -def test_update_production_model(mlflow_repository): - connector = mlflow_repository - - with patch.object(connector, 'get_experiment', - return_value='0') as get_experiment: - with patch.object(connector, 'get_experiment_last_run', - return_value='2') as get_experiment_last_run: - with patch.object(connector, 'update_production_model_by_run_id', - return_value={'model_name': 'test', 'version': '3', - 'mlflow_run_id': '0'}) as update_production_model_by_run_id: - - output = connector.update_production_model('0', 'test') - - get_experiment.assert_called_once_with('0') - get_experiment_last_run.assert_called_once_with('0') - update_production_model_by_run_id.assert_called_once_with( - '2', 'test') - - assert output == { - 'model_name': 'test', - 'version': '3', - 'mlflow_run_id': '0', - 'mlflow_experiment_id': '0', - } - - def test_transform_success(mlflow_repository): data = 'data' model_name = 'model' diff --git a/tests/laborious/utils/test_logger.py b/tests/laborious/utils/test_logger.py deleted file mode 100644 index cb68cb4..0000000 --- a/tests/laborious/utils/test_logger.py +++ /dev/null @@ -1,37 +0,0 @@ -import os -from unittest.mock import patch -import logging -import pytest -from laborious.utils.logger import get_logger - - -@pytest.fixture -def mock_env_vars(): - with patch.dict(os.environ, {}, clear=True): - yield - - -@pytest.mark.usefixtures("mock_env_vars") -@patch('laborious.utils.logger.logging.Formatter') -@patch('laborious.utils.logger.logging.StreamHandler') -def test_get_logger_defaults(mock_stream_handler, mock_formatter): - """Test logger creation with default settings""" - # Mock the StreamHandler and Formatter - - logger = get_logger('test_logger') - - # Verify logger settings - assert logger.name == 'test_logger' - assert logger.level == logging.INFO - - # Verify handler configuration - mock_stream_handler.return_value.setLevel.assert_called_once_with('INFO') - mock_stream_handler.return_value.setFormatter.assert_called_once() - - # Verify formatter configuration - mock_formatter.assert_called_once_with( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - ) - - # Verify handler was added to logger - assert len(logger.handlers) == 1 From 10081b71d2ad536e216f55bbd72dc5c9ebc82180 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 26 May 2025 16:45:50 -0300 Subject: [PATCH 8/9] SIENTIAPDE-1081 Enhance documentation across multiple modules with detailed parameter descriptions and usage examples --- README.md | 95 +++++++++++++++++++ laborious/activities/gates.py | 58 +++++------ laborious/activities/mlflow.py | 16 ++-- laborious/activities/opc.py | 22 +++-- laborious/utils/connectors_config.py | 4 + .../utils/filters/conditional_filters.py | 14 +++ laborious/utils/filters/mlflow_filters.py | 20 ++++ .../utils/repository/model_repository.py | 31 +++++- laborious/utils/repository/opc_repository.py | 53 +++++++++-- laborious/workflows/predictions_batch.py | 25 ++--- .../format_and_export_prediction.py | 22 ++--- .../sub_workflows/prediction_process.py | 26 ++--- 12 files changed, 294 insertions(+), 92 deletions(-) diff --git a/README.md b/README.md index e69de29..d69a75c 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,95 @@ +# Sientia DataOps Laborious + +The Sientia DataOps Laborious is a Temporal-based workflow application that handles batch predictions and data processing for industrial data. It integrates with MLFlow for model management, PostgreSQL for data storage, and OPC for real-time data output. The module is designed to process data in a reliable and scalable manner using Temporal.io's workflow orchestration capabilities. It's get data from Scouter sinks, process it, make predictions using MLFlow models and generates metrics for the predictions. + +## Key Features + +- Batch predictions using MLFlow models +- Data transformation and preprocessing +- Workflow orchestration using Temporal.io +- Integration with PostgreSQL for data storage +- OPC integration for real-time data output +- Comprehensive error handling and notifications +- Configurable data filters and quality gates +- Scalable deployment architecture + +## Workflows + +### Predictions Batch +The main workflow that orchestrates batch predictions. Steps: + +- prepare_activity: Prepares the activity with schedule and model information +- load_custom_query: Loads data using a custom query +- prediction_process: Executes the prediction process using the Prediction Process sub-workflow + +#### Workflow inputs: + +- `schedule_name`: The schedule name of the activity +- `model_name`: The model name of the activity +- `model_id`: The model id of the activity +- `query`: The custom query to load data +- `schema`: The schema of the data +- `table_name`: The name of the table to process +- `input_filters`: The filters to be applied during prediction +- `mlflow_transform_filters`: The filters to be applied during prediction +- `mlflow_predict_filters`: The filters to be applied during prediction +- `model_retention`: The model retention period in minutes +- `path_priority`: The path priority + + +### Prediction Process +Sub-workflow that handles individual prediction processing: + +- get_last_timestamp: Gets the last timestamp of the data +- input_gate: Filters input data based on configured rules +- repeat_last_prediction: Repeats the last prediction if the data is empty +- request_transform: Makes predictions using MLFlow models +- mlflow_response_gate: Handles prediction or transform responses and filters +- mlflow_content_gate: Filters transform responses based on configured rules +- request_predict: Makes predictions using MLFlow models +- format_and_export_prediction: Formats and exports predictions using the + Format and Export Prediction sub-workflow + +### Format and Export Prediction +Sub-workflow that handles prediction formatting and export: + +- format_prediction: Formats prediction data if path flag is None +- format_default_prediction: Formats default prediction data if path flag is not None +- export_to_postgres: Exports formatted predictions to PostgreSQL +- write_to_opc: Writes predictions to OPC server + +## Environment variables + +- `POSTGRES_HOST` +- `POSTGRES_PORT` +- `POSTGRES_USER` +- `POSTGRES_PASSWORD` +- `POSTGRES_DBNAME` +- `POSTGRES_MIN_CONNECTIONS` +- `POSTGRES_MAX_CONNECTIONS` + +- `MLFLOW_HOST` +- `MLFLOW_PORT` +- `MLFLOW_USERNAME` +- `MLFLOW_PASSWORD` + +- `OPC_CONFIG` - json string containing the opc configuration for multiple opc servers +For single opc server use: +- `OPC_URL` +- `OPC_NAME` +- `OPC_SERVER_URI` +- `OPC_CERT_PATH` +- `OPC_PRIVATE_KEY_PATH` +- `OPC_SERVER_CERT_PATH` +- `OPC_RECONNECTION_INTERVAL` + +- `TEMPORAL_HOST` +- `TEMPORAL_NAMESPACE` + +## Application deployment + +The application can be deployed using the following command: + +```bash +helm upgrade --install sientia-dataops-laborious sientia/sientia-module -n sientia --create-namespace -f ./values.yaml +``` diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 799fed5..f0796ff 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -55,13 +55,13 @@ class Gates(BaseActivity): Filters the data based on the filters. The return value is a tuple with the first element being the policy and the second element being the confidence status. Args: - input_data (dict): The input data. Contains: - filters (dict): The filters to apply. + - input_data (dict): The input data. Contains: + - filters (dict): The filters to apply. The key is the filter name and the value is the filter configuration. - data (dict[str, Any]): The data to filter. - path_priority (list[str]): The path priority. + - data (dict[str, Any]): The data to filter. + - path_priority (list[str]): The path priority. Returns: - tuple[str | None, int, str]: (policy, confidence) based in priority + tuple[str | None, int, str]: (policy, confidence, comments) based in priority list and filter configuration and functions. """ @@ -111,13 +111,13 @@ class Gates(BaseActivity): The return value is a tuple with the first element being the policy and the second element being the confidence status. Args: - input_data (dict): The input data. Contains: - filters (dict): The filter configuration to apply. - data (dict[str, Any]): The data to filter. - path_priority (list[str]): The path priority list. - type (str): The type of the gate. + - input_data (dict): The input data. Contains: + - filters (dict): The filter configuration to apply. + - data (dict[str, Any]): The data to filter. + - path_priority (list[str]): The path priority list. + - type (str): The type of the gate. Returns: - tuple[str | None, int, str]: (policy, confidence) based in priority list + tuple[str | None, int, str]: (policy, confidence, comments) based in priority list and filter configuration and functions. """ @@ -174,13 +174,13 @@ class Gates(BaseActivity): The return value is a tuple with the first element being the policy and the second element being the confidence status. Args: - input_data (dict): The input data. Contains: - filters (dict): The filter configuration to apply. - data (dict[str, Any]): The data to filter. - path_priority (list[str]): The path priority list. - type (str): The type of the gate. + - input_data (dict): The input data. Contains: + - filters (dict): The filter configuration to apply. + - data (dict[str, Any]): The data to filter. + - path_priority (list[str]): The path priority list. + - type (str): The type of the gate. Returns: - tuple[str | None, int, str]: (policy, confidence) based in priority + tuple[str | None, int, str]: (policy, confidence, comments) based in priority list and filter configuration and functions. """ @@ -233,11 +233,11 @@ class Gates(BaseActivity): """ Formats the prediction data. Args: - input_data (dict): The input data. Contains: - data (dict[str, Any]): The data to format. - timestamp (str): The timestamp of the data. - model_id (str): The id of the model. - prediction_confidence (float): The confidence of the prediction. + - input_data (dict): The input data. Contains: + - data (dict[str, Any]): The data to format. + - timestamp (str): The timestamp of the data. + - model_id (str): The id of the model. + - prediction_confidence (float): The confidence of the prediction. Returns: dict: The formatted data. """ @@ -260,11 +260,11 @@ class Gates(BaseActivity): and usefull information in the other fields. Args: - input_data (dict): The input data. Contains: - timestamp (str): The timestamp of the data. - model_id (str): The id of the model. - prediction_confidence (float): The confidence of the prediction. - comment (str): The comment of the prediction. + - input_data (dict): The input data. Contains: + - timestamp (str): The timestamp of the data. + - model_id (str): The id of the model. + - prediction_confidence (float): The confidence of the prediction. + - comment (str): The comment of the prediction. Returns: dict: The formatted data. """ @@ -286,8 +286,8 @@ class Gates(BaseActivity): """ Gets the last timestamp of the data. Args: - input_data (dict): The input data. Contains: - data (dict[str, Any]): The data to get the last timestamp from. + - input_data (dict): The input data. Contains: + - data (dict[str, Any]): The data to get the last timestamp from. Returns: str: The last timestamp of the data. """ diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index a209ed1..9b8a58a 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -29,10 +29,10 @@ class MLFlow(BaseActivity): """ Access MLFlow model to get the transformed data. Args: - input_data (dict): The input data. Contains: - data (dict[str, Any]): The data to transform. - model_name (str): The name of the model. - model_retention (int): The retention of the model in minutes. + - input_data (dict): The input data. Contains: + - data (dict[str, Any]): The data to transform. + - model_name (str): The name of the model. + - model_retention (int): The retention time of the model, in minutes. Returns: dict[str, Any]: The transformed data. """ @@ -67,10 +67,10 @@ class MLFlow(BaseActivity): """ Access MLFlow model to get the predicted data. Args: - input_data (dict): The input data. Contains: - data (dict[str, Any]): The data to predict. - model_name (str): The name of the model. - model_retention (int): The retention of the model. + - input_data (dict): The input data. Contains: + - data (dict[str, Any]): The data to predict. + - model_name (str): The name of the model. + - model_retention (int): The retention time of the model, in minutes. Returns: dict[str, Any]: The predicted data. """ diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 997b7c2..11cd77c 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -39,6 +39,17 @@ class OPC(BaseActivity): def write_data(self, server: str, tag: str, data: Any, data_type: str, tag_type: str): + """ + Write data to OPC server. + + Args: + - server (str): The name of the OPC server. + - tag (str): The tag to write to. + - data (Any): The data to write. + - data_type (str): The data type. + - tag_type (str): The tag type. + """ + try: self.opc_repository[server].write_data( tag, data, data_type) @@ -61,15 +72,14 @@ class OPC(BaseActivity): operations are optional and independent of each other. Args: - input_data (dict[str, Any]): The input data. Contains the following keys: - - data (dict[str, Any]): The dataframe that contains the data to write + - input_data(dict[str, Any]): The input data. Contains the following keys: + - data(dict[str, Any]): The dataframe that contains the data to write to the OPC servers. - - opc_output_config (dict[str, Any]): The OPC writing configuration. + - opc_output_config(dict[str, Any]): The OPC writing configuration. The keys are the OPC server names and the values contain: - prediction_tags (dict[str, Any]): The tags to write to the OPC servers. - confidence_tags (dict[str, Any]): The tags to write to the OPC servers. + - prediction_tags(dict[str, Any]): The tags to write to the OPC servers. + - confidence_tags(dict[str, Any]): The tags to write to the OPC servers. - Returns: """ self.logger.debug("Writing data to OPC servers...") data = DataFrame(input_data['data']) diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 80ed24d..0515cc2 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -1,3 +1,7 @@ +""" +Builds the configuration for the connectors. +""" + from os import getenv import json diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index 57cd3fd..d98cebe 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -4,6 +4,13 @@ from pandas import DataFrame def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool: """ Returns True if the specific columns have null values, False otherwise. + + Args: + - data (DataFrame): The data to filter. + - config (dict): The configuration. + + Returns: + bool: True if the specific columns have null values, False otherwise. """ return not data[ data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty @@ -12,5 +19,12 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool def filter_empty_data(data: DataFrame, _config: dict) -> bool: """ Returns True if the data is empty, False otherwise. + + Args: + - data (DataFrame): The data to filter. + - _config (dict): The configuration. + + Returns: + bool: True if the data is empty, False otherwise. """ return data.empty diff --git a/laborious/utils/filters/mlflow_filters.py b/laborious/utils/filters/mlflow_filters.py index 9936018..f6f1efc 100644 --- a/laborious/utils/filters/mlflow_filters.py +++ b/laborious/utils/filters/mlflow_filters.py @@ -3,6 +3,16 @@ from pandas import DataFrame def api_error_filter(response: dict, _config: dict): + """ + Returns True if the API response is empty or the 'success' key is False, False otherwise. + + Args: + - response (dict): The API response. + - _config (dict): The configuration. + + Returns: + bool: True if the API response is empty or the 'success' key is False, False otherwise. + """ if not response: return True @@ -13,6 +23,16 @@ def api_error_filter(response: dict, _config: dict): def nan_values_filter(predictions: DataFrame, _config: dict): + """ + Returns True if the predictions DataFrame contains only NaN values, False otherwise. + + Args: + - predictions (DataFrame): The predictions DataFrame. + - _config (dict): The configuration. + + Returns: + bool: True if the predictions DataFrame contains only NaN values, False otherwise. + """ data = predictions.replace({None: np.nan}).drop( columns=['timestamp'], errors='ignore').infer_objects(copy=False) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index ed5936b..b3473c6 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -1,11 +1,11 @@ """ Model Monitoring Repository -This module contains the ModelMonitoringRepository class, which is responsible -for handling the communication with the Model Monitoring API. +This module contains the ModelMonitoringRepository class, +which is responsible for handling the communication with the Model Monitoring API. -It includes the methods that are used to answer ModelMonitoringService requests using -the Model Monitoring API functions. +It includes the methods that are used to answer ModelMonitoringService +requests using the Model Monitoring API functions. By Monitoring we mean the evaluation of the performance of models, the generation of reports. @@ -23,6 +23,18 @@ class MLFlowRepository(): username=username, password=password) def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): + """ + Transform data using a model. + + Parameters: + - model_name (str): The name of the model to use for transformation. + - data (pandas.DataFrame): The data to transform. + - model_retention (int): The number of minutes to keep the model. + + Returns: + - dict: A dictionary containing the transformed data. + """ + try: return { 'success': True, @@ -40,6 +52,17 @@ class MLFlowRepository(): } def predict(self, model_name: str, data: pd.DataFrame, model_retention: int): + """ + Predict data using a model. + + Parameters: + - model_name (str): The name of the model to use for prediction. + - data (pandas.DataFrame): The data to predict. + - model_retention (int): The number of minutes to keep the model. + + Returns: + - dict: A dictionary containing the predicted data. + """ try: start_time = datetime.now() data = self.model_serving.get_cached_predict( diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index e674354..cafcc41 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,12 +1,12 @@ +import traceback +from logging import Logger +from datetime import datetime from pathlib import Path from asyncua.sync import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.ua import DataValue, Variant, VariantType -from logging import Logger -from datetime import datetime from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel -import traceback data_type_map = { 'float': { @@ -33,7 +33,8 @@ data_type_map = { class OpcRepository(): - def __init__(self, name: str, url: str, logger: Logger, notification_handler: NotificationHandler, + def __init__(self, name: str, url: str, logger: Logger, + notification_handler: NotificationHandler, reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): self.url = url @@ -57,12 +58,12 @@ class OpcRepository(): Raises: ValueError: If either the certificate path or private key path is not provided. Attributes: - cert_path (str): Path to the client's certificate file. - private_key_path (str): Path to the client's private key file. - server_cert_path (str, optional): Path to the server's certificate file. - server_uri (str): The URI of the server to be used as the application URI. - client (opcua.Client): The OPC UA client instance. - logger (logging.Logger): Logger instance for logging information. + - cert_path (str): Path to the client's certificate file. + - private_key_path (str): Path to the client's private key file. + - server_cert_path (str, optional): Path to the server's certificate file. + - server_uri (str): The URI of the server to be used as the application URI. + - client (opcua.Client): The OPC UA client instance. + - logger (logging.Logger): Logger instance for logging information. Security Settings: - Security Policy: Basic256 - Secure Channel Timeout: 10,000,000 ms @@ -105,6 +106,12 @@ class OpcRepository(): return self.try_connect() def try_connect(self): + """ + Tries to connect to the OPC server. + + Returns: + bool: True if the connection was successful, False otherwise. + """ try: self.last_reconnection_time = datetime.now() self.client.connect() @@ -122,6 +129,9 @@ class OpcRepository(): return False def disconnect(self): + """ + Disconnects from the OPC server. + """ if self.client is None: return self.client.disconnect() @@ -129,12 +139,25 @@ class OpcRepository(): self.logger.info('Disconnected from OPC server') def __del__(self): + """ + Disconnects from the OPC server when the object is destroyed. + """ try: self.disconnect() except Exception as e: self.logger.error(f"Error in destructor: {e}") def validate_connection(self): + """ + Validates the connection to the OPC server. + If the connection is not established, it attempts to reconnect. + If the connection is established but the client is not connected, + it attempts to reconnect. + If the connection is established but the client is connected, + it checks if the client is connected to the OPC server. + If the client is not connected, it attempts to reconnect. + If the client is connected, it returns True. + """ if self.client is None: return self.connect() @@ -168,6 +191,16 @@ class OpcRepository(): return True def write_data(self, node, value, data_type): + """ + Writes data to the OPC server. + If the connection is not established, it attempts to reconnect. + If the connection is established but the client is not connected, + it attempts to reconnect. + If the connection is established but the client is connected, + it checks if the client is connected to the OPC server. + If the client is not connected, it attempts to reconnect. + If the client is connected, it returns True. + """ if not self.validate_connection(): return try: diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 9c4ab61..d80a9d4 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -19,19 +19,20 @@ class PredictionsBatch(): 2. Loads data using a custom query and executes the prediction process Args: - input_data (dict[str, Any]): The input data for the workflow. + - input_data (dict[str, Any]): The input data for the workflow. Contains the following keys: - schedule_name (str): The name of the schedule. - model_name (str): The name of the model. - model_id (int): The id of the model. - query (str): The SQL query to be executed to load data. - schema (dict, optional): The schema definition for the data. - table_name (str, optional): The name of the table to process. - input_filters (dict, optional): Filters to be applied during prediction. - mlflow_transform_filters (dict, optional): Filters to be applied during prediction. - mlflow_predict_filters (dict, optional): Filters to be applied during prediction. - model_retention (int, optional): The model retention period in minutes. - path_priority (list[str]): The path priority. + - schedule_name (str): The name of the schedule. + - model_name (str): The name of the model. + - model_id (int): The id of the model. + - query (str): The SQL query to be executed to load data. + - schema (dict, optional): The schema definition for the data. + - table_name (str, optional): The name of the table to process. + - input_filters (dict, optional): Filters to be applied during prediction. + - mlflow_transform_filters (dict, optional): Filters to be applied + during prediction. + - mlflow_predict_filters (dict, optional): Filters to be applied during prediction. + - model_retention (int, optional): The model retention period in minutes. + - path_priority (list[str]): The path priority. Returns: None diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 68a5958..d904fee 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -21,17 +21,17 @@ class FormatAndExportPrediction(): Args: input_data(dict[str, Any]): The input data for the workflow. Contains the following keys: - path_flag(str): The path flag to determine the type of prediction to format - data(dict[str, Any]): The data to format - prediction_confidence(float): The prediction confidence to be registered - timestamp(str): The timestamp of the prediction, synchronized with the data - model_id(int): The model id of the prediction - model_name(str): The model name of the prediction - model_retention(str): The model retention of the prediction - comment(str): The comment to be registered - schema(str): The schema of the prediction - table_name(str): The table name of the prediction - opc_output_config(dict[str, Any]): The opc output config of the prediction + - path_flag(str): The path flag to determine the type of prediction to format + - data(dict[str, Any]): The data to format + - prediction_confidence(float): The prediction confidence to be registered + - timestamp(str): The timestamp of the prediction, synchronized with the data + - model_id(int): The model id of the prediction + - model_name(str): The model name of the prediction + - model_retention(str): The model retention of the prediction + - comment(str): The comment to be registered + - schema(str): The schema of the prediction + - table_name(str): The table name of the prediction + - opc_output_config(dict[str, Any]): The opc output config of the prediction Returns: bool: True if the workflow was successful, False otherwise. diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index df740b3..e4fc8cc 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -19,19 +19,21 @@ class PredictionProcess(): 2. Loads data using a custom query and executes the prediction process Args: - input_data (dict[str, Any]): The input data for the workflow. + - input_data (dict[str, Any]): The input data for the workflow. Contains the following keys: - data (dict[str, Any]): The data to be used for the prediction. - schema (str): The schema of the table. - table_name (str): The name of the table. - model_id (int): The id of the model. - input_filters (dict, optional): Filters to be applied during prediction. - mlflow_transform_filters (dict, optional): Filters to be applied during prediction. - mlflow_predict_filters (dict, optional): Filters to be applied during prediction. - model_name (str): The name of the model. - model_retention (int, optional): The model retention period in minutes. - path_priority (list[str]): The path priority. - opc_output_config (dict[str, Any]): The opc output config of the prediction. + - data (dict[str, Any]): The data to be used for the prediction. + - schema (str): The schema of the table. + - table_name (str): The name of the table. + - model_id (int): The id of the model. + - input_filters (dict, optional): Filters to be applied during prediction. + - mlflow_transform_filters (dict, optional): Filters to be + applied during prediction. + - mlflow_predict_filters (dict, optional): Filters to be + applied during prediction. + - model_name (str): The name of the model. + - model_retention (int, optional): The model retention period in minutes. + - path_priority (list[str]): The path priority. + - opc_output_config (dict[str, Any]): The opc output config of the prediction. Returns: None From 497e7a2eaa1c798574d6abd0a52e555769979a00 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 9 Jun 2025 12:44:39 -0300 Subject: [PATCH 9/9] conflicts --- .../utils/repository/model_repository.py | 244 ------------------ 1 file changed, 244 deletions(-) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 5478103..b3473c6 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -22,250 +22,6 @@ class MLFlowRepository(): self.model_serving = ModelServing(tracking_uri=host, username=username, password=password) -<<<<<<< HEAD -======= - def get_current_data_df(self, current_data: pd.DataFrame, model_name: str, target: str): - """ - Get the current data as a DataFrame and update the prediction and target columns - - Parameters: - current_data (pd.DataFrame): the current data - model_name (str): the name of the model - target (str): the target column - - Returns: - DataFrame: the current data as a DataFrame - - - """ - predictions = current_data['prediction'] - - target = current_data[target] - current_data = self.model_serving.get_transformed_data( - model_name, current_data, by='model') - current_data['prediction'] = predictions - current_data['target'] = target - - return pd.DataFrame(current_data).dropna() - - def get_artifact(self, destination: str, search_by: str, run_id: str = None, - model_name: str = None, artifact_name: str = None) -> None: - """ - Get an artifact in MLflow by experiment or model and save it to a - destination path using API. - If the artifact is searched by model, the latest production version will be used. - - Args: - destination: The destination path to save the artifact. - search_by: The way to search for the artifact ('experiment' or 'model'). - run_id: The run ID of the experiment (if search_by is "experiment"). - model_name: The name of the model (if search_by is "model"). - artifact_name: The path of the artifact to download. - - Returns: - artifact: The artifact(.csv) downloaded from MLflow. - """ - - self.model_serving.get_artifact(destination=destination, search_by=search_by, - run_id=run_id, model_name=model_name, artifact_name=artifact_name) - - def calculate_model_metrics(self, real_data, predictions, flag): - """ - Function to calculate the metrics of a model using API - - Parameters: - real_data (array): the real data - predictions (array): the predictions - - Returns: - dict: the metrics of the model including MSE and R2 - """ - return self.model_serving.get_model_metrics( - reference_data=None, real_data=real_data, - predictions=predictions, type_flag=flag) - - def get_experiment_by_run_id(self, run_id: str) -> dict: - # Get the run information using the run_id - run = mlflow.get_run(run_id) - - # Extract the experiment ID from the run - experiment_id = run.info.experiment_id - - # Get the experiment details using the experiment ID - experiment = mlflow.get_experiment(experiment_id) - experiment_name = experiment.name - return experiment_name - - def get_next_run_name(self, model_name: str) -> str: - """ - Function to get the next run number of a specific model - - Parameters: - model_name (str): the name of the model - - Returns: - str: the next run number - """ - - runs = mlflow.search_runs( - experiment_names=[model_name], order_by=["start_time desc"]) - next_run_number = len(runs) + 1 - return f"{model_name}-{next_run_number}" - - def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple: - """ - Retrain a model with new data. - - Parameters: - - data (pandas.DataFrame): The new data to use for retraining. - - model_name (str): The name of the model to retrain. - - metrics_list (list): The metrics to be used to compare the models. - - compare_metrics (bool): If True, the retrain will only be considered if - the new model is better than the current one. - If False, the retrain will always be considered. - - split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets. - If False, the data will be used as a unique block for retraining. - - update_report (bool): If True, a report will be created with the data of the retrained model. - - update_transformation (bool): If True, the model will be updated in the MLflow tracking server. - - update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server. - - shuffle_data (bool): If True, the data will be shuffled before splitting. - - model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'. - - - Returns: - - mlflow.sklearn.Model: The retrained prediction model. - - mlflow.sklearn.Model: The retrained data model. - - mse (float): The mean squared error of the retrained model. - - r2 (float): The R-squared score of the retrained model. - """ - - # load predictor model - predictor_uri = f"models:/{model_name}/production" - # load transform model - latest_production_id = self.model_serving.get_model_run_id( - model_name, stage="Production" - ) - transform_uri = self.model_serving.get_model_uri( - latest_production_id, prediction=False - ) - # load - data_model = mlflow.sklearn.load_model(transform_uri) - prediction_model = mlflow.sklearn.load_model(predictor_uri) - data_model = data_model.fit(data) - treated_data = data_model.predict(data) - # align target column with treated_data - target_name = data_model.target_variable - y = data[target_name] - treated_data = pd.merge( - treated_data, y, left_index=True, right_index=True) - prediction_model = prediction_model.fit(treated_data) - # Example usage - experiment = self.get_experiment_by_run_id(latest_production_id) - pred_model_atributes = vars(prediction_model) # load class attributes - data_model_atributes = vars(data_model) # load class attributes - mlflow.set_experiment(experiment) - experiment_description = "Retrain model {model_name} with new data" - current_run_name = self.get_next_run_name(experiment) - with mlflow.start_run( - run_name=current_run_name, description=experiment_description - ) as _run: - # update transfomation model - # fixed parameters - for name_atribute, val_atribute in pred_model_atributes.items(): - if name_atribute != "model": - mlflow.log_param(name_atribute, val_atribute) - # update prediction model - for name_atribute, val_atribute in data_model_atributes.items(): - if name_atribute != "model": - mlflow.log_param(name_atribute, val_atribute) - # dynamic parameters, including model itself - mlflow.sklearn.log_model(data_model, "data_model") - file_path = f"laborious/data/raw_data_{model_name}.csv" - data.to_csv( - f"laborious/data/raw_data_{model_name}.csv", index=True) - # log the data raw - mlflow.log_artifact(file_path) - - # dynamic parameters, including model itself - mlflow.sklearn.log_model(prediction_model, "prediction_model") - mlflow.log_param("retrain", True) - - return "Model retrained successfully", experiment - - def get_experiment(self, experiment_name: str) -> int: - experiment = mlflow.get_experiment_by_name(experiment_name) - - if experiment is None: - raise ValueError(f'Experiment {experiment_name} not found') - - return int(experiment.experiment_id) - - def get_experiment_last_run(self, experiment_id: int) -> str: - runs = mlflow.search_runs( - experiment_ids=[experiment_id], - filter_string="", # Sem filtro no MLflow ainda - output_format="pandas" - ) - - # Filtrar apenas as runs onde params.retrain == True - filtered_runs = runs[runs["params.retrain"] == 'True'] - - # Converter a coluna 'end_time' para datetime - filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time']) - - # Ordenar o DataFrame de forma descendente pela coluna 'end_time' - filtered_runs = filtered_runs.sort_values( - by='end_time', ascending=False) - - # Pegar a última run_id do DataFrame filtrado e ordenado - latest_run_id = filtered_runs.iloc[0]['run_id'] - - return latest_run_id - - def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict: - # Registrar o modelo - # Aqui estamos assumindo que você já tem um modelo salvo, - # caso contrário você precisará treiná-lo e salvá-lo primeiro. - # Se o modelo já está registrado, você pode usar o método - # register_model() ou pyfunc.load_model() para isso. - mlflow.register_model( - f"runs:/{run_id}/prediction_model", model_name) - - # Colocar a versão do modelo em produção - # Depois de registrar o modelo, precisamos pegar a versão mais - # recente do modelo e movê-lo para o estágio 'Production' - client = mlflow.tracking.MlflowClient() - - # Obter a versão mais recente registrada do modelo - model_versions = client.get_registered_model( - model_name).latest_versions - max_version = max(model_versions, key=lambda x: int(x.version)).version - - # Mover a versão mais recente do modelo para o estágio de 'Production' - client.transition_model_version_stage( - name=model_name, - version=max_version, - stage="Production", - archive_existing_versions=True - ) - - return { - 'model_name': model_name, - 'version': max_version, - 'mlflow_run_id': run_id - } - - def update_production_model(self, experiment: str, model_name: str) -> dict: - - experiment_id = self.get_experiment(experiment) - run_id = self.get_experiment_last_run(experiment_id) - metadata = self.update_production_model_by_run_id(run_id, model_name) - - metadata['mlflow_experiment_id'] = experiment_id - - return metadata - ->>>>>>> main def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): """ Transform data using a model.