diff --git a/laborious/activities/api.py b/laborious/activities/api.py index 063fd79..7d70684 100644 --- a/laborious/activities/api.py +++ b/laborious/activities/api.py @@ -148,7 +148,10 @@ class API(SientiaMonitoring): if len(written_tags) != len(tag_names): message = f'The number of written tags does not match the number of tag names: Expected {tag_names} tags, but {written_tags} tags were written.' - self.error(f"{message}\nResponse:\n {json.dumps(response_data, indent=4)}\nTags:\n {json.dumps(tags, indent=4)}", metadata) + self.error( + f'{message}\nResponse:\n {json.dumps(response_data, indent=4)}\nTags:\n {json.dumps(tags, indent=4)}', + metadata, + ) await self.send_notification_async( metadata=metadata, diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 4b5eec8..b581920 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -6,13 +6,13 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from pandas import DataFrame - from sientia_do.utils.formatters import create_sample_dict from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now + from sientia_do.utils.formatters import create_sample_dict from laborious import metrics from laborious.utils.filters.conditional_filters import ( diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 4a51d5e..082e720 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -6,7 +6,6 @@ with workflow.unsafe.imports_passed_through(): import numpy as np from pandas import DataFrame, to_datetime - from sientia_do.utils.formatters import create_sample_dict from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -18,6 +17,7 @@ with workflow.unsafe.imports_passed_through(): DATETIME_FORMAT_WITH_TZ, now, ) + from sientia_do.utils.formatters import create_sample_dict from laborious.utils.repository.minio_repository import MinioRepository from laborious.utils.repository.model_repository import MLFlowRepository @@ -159,7 +159,6 @@ class MLFlow(SientiaMonitoring): self.debug(f'Processed input data: \n {data.to_csv()}', metadata) - # Request transformation from MLFlow model response_data = await self.model_monitoring_repository.transform( model_name, data, model_config, metadata diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 84bbbf0..66862e9 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -1254,7 +1254,9 @@ class MLFlowRepository(SientiaMonitoring): input_index = data.index start_time = datetime.now() - self.debug(f'Data received for model prediction: {data.to_dict(orient="records")}', metadata) + self.debug( + f'Data received for model prediction: {data.to_dict(orient="records")}', metadata + ) # data.to_csv( # f"tmp/treated_data_{model_name}.csv", index=True) @@ -1272,7 +1274,8 @@ class MLFlowRepository(SientiaMonitoring): if isinstance(predict_data, pd.DataFrame): self.debug( - f'Data received from model prediction: {predict_data.to_dict(orient="records")}', metadata + f'Data received from model prediction: {predict_data.to_dict(orient="records")}', + metadata, ) # predict_data.to_csv( diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index faf271a..b5210d4 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -31,27 +31,20 @@ Environment Variables: from temporalio import client, workflow from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig -from temporalio.worker import ( - PollerBehaviorAutoscaling, - ResourceBasedSlotConfig, - Worker, - WorkerTuner, -) with workflow.unsafe.imports_passed_through(): import asyncio import os import sys - from datetime import timedelta from prometheus_client import start_http_server + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.observability.logger import get_logger from sientia_do.utils.connectors_config import ( build_api_config, build_mongodb_config, build_postgres_config, ) - from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler - from sientia_do.observability.logger import get_logger from laborious import metrics from laborious.activities.activities import Activities @@ -60,6 +53,7 @@ with workflow.unsafe.imports_passed_through(): build_mlflow_config, build_opc_config, ) + from laborious.worker.prepare_worker import prepare_worker from laborious.workflows.drift import Drift from laborious.workflows.minimal_retrain import MinimalRetrain from laborious.workflows.predictions_batch import PredictionsBatch @@ -68,11 +62,11 @@ with workflow.unsafe.imports_passed_through(): FormatAndExportPrediction, ) from laborious.workflows.sub_workflows.prediction_process import PredictionProcess - from laborious.worker.prepare_worker import prepare_worker POD_ID = os.getenv('POD_ID') SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) + async def main(): """ Main entry point for the Laborious worker application. @@ -215,7 +209,7 @@ async def main(): activities.write_metrics, ], logger=logger, - ) + ), ] handlers = [] diff --git a/tests/laborious/activities/test_api.py b/tests/laborious/activities/test_api.py index ad9866f..5226e3e 100644 --- a/tests/laborious/activities/test_api.py +++ b/tests/laborious/activities/test_api.py @@ -333,11 +333,13 @@ async def test_process_pi_web_api_response_with_errors(api): ) assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE - assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written." + assert ( + message + == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written." + ) assert api.emit_metric.call_count == 2 - @mark.asyncio async def test_process_pi_web_api_response_missing_tags(api): """Test processing response when number of written tags doesn't match expected.""" @@ -361,7 +363,10 @@ async def test_process_pi_web_api_response_missing_tags(api): ) assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE - assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag1'] tags were written." + assert ( + message + == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag1'] tags were written." + ) api.send_notification_async.assert_called_once() call_args = api.send_notification_async.call_args assert call_args.kwargs['notification_id'] == 'WRITE_PI_WEB_API_PREDICTION_ERROR' @@ -392,7 +397,10 @@ async def test_process_pi_web_api_response_missing_webid(api): ) assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE - assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written." + assert ( + message + == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written." + ) api.error.assert_any_call('The response did not contain some WebIds', metadata['metadata']) @@ -419,7 +427,10 @@ async def test_process_pi_web_api_response_missing_tag_name(api): ) assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE - assert message == "The number of written tags does not match the number of tag names: Expected ['tag1'] tags, but [] tags were written." + assert ( + message + == "The number of written tags does not match the number of tag names: Expected ['tag1'] tags, but [] tags were written." + ) api.error.assert_any_call( 'The response did not contain the tag name for WebId unknown_web_id', metadata['metadata'] ) diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index d3f98b6..0bd09d1 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -812,7 +812,14 @@ async def test_fit_models_not_df_target_name_none_and_not_in_model( data = MagicMock() output = await mlflow_repository.fit_models( - 'model_name', data, 'latest_production_id', metadata['metadata'], 'sklearn', False, 'pyfunc', None + 'model_name', + data, + 'latest_production_id', + metadata['metadata'], + 'sklearn', + False, + 'pyfunc', + None, ) mlflow_repository.download_model.assert_has_calls( diff --git a/validate.sh b/validate.sh index 87a8df8..6c72694 100755 --- a/validate.sh +++ b/validate.sh @@ -70,14 +70,14 @@ FAILED_STEPS=() # Step 1: Code Formatting Check (Ruff) # - default: check only # - --fix: write changes -if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format laborious/ tests/; else ruff format --check laborious/ tests/; fi"; then +if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format laborious/ tests/; else ruff format --check laborious/ tests/ e2e/; fi"; then FAILED_STEPS+=("Code Formatting") fi # Step 2: Linting (Ruff) # - default: check only # - --fix: apply autofixes -if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix laborious/ tests/; else ruff check laborious/ tests/; fi"; then +if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix laborious/ tests/; else ruff check laborious/ tests/ e2e/; fi"; then FAILED_STEPS+=("Linting") fi