67fe4afaa6
Refactor and enhance the laborious workflow and utilities - Removed outdated test file `test_predictions_batch.py` from workflows. - Added `input_sample.json` for standardized input configuration. - Introduced `connectors_config.py` to manage database and service configurations. - Implemented a logging utility in `logger.py` for consistent logging across the application. - Created `policies.py` to define retry policies for workflows. - Developed comprehensive tests for `MLFlowRepository` in `test_model_repository.py`. - Added extensive tests for `OpcRepository` in `test_opc_repository.py`. - Updated `test_predictions_batch.py` to reflect new workflow structure and testing methodology.
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import numpy as np
|
|
from pandas import DataFrame
|
|
from temporalio import activity, workflow
|
|
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from laborious.activities.base import BaseActivity
|
|
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):
|
|
def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str,
|
|
mlflow_password: str, logger: Logger, notification_handler: NotificationHandler):
|
|
BaseActivity.__init__(self, logger, notification_handler)
|
|
self.mlflow_host = mlflow_host
|
|
self.mlflow_port = mlflow_port
|
|
self.mlflow_username = mlflow_username
|
|
self.mlflow_password = mlflow_password
|
|
|
|
self.model_monitoring_repository = MLFlowRepository(
|
|
f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password
|
|
)
|
|
|
|
@activity.defn(name="request_transform")
|
|
async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
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.
|
|
Returns:
|
|
dict[str, Any]: The transformed data.
|
|
"""
|
|
self.logger.info('Transforming data...')
|
|
data = DataFrame(input_data['data'])
|
|
model_name = input_data['model_name']
|
|
model_retention = input_data['model_retention']
|
|
|
|
self.logger.debug(data)
|
|
|
|
data = data.pivot(
|
|
index='timestamp', columns='variable',
|
|
values='value')
|
|
data.fillna(np.nan, inplace=True)
|
|
data.reset_index(inplace=True)
|
|
data.columns.name = None
|
|
|
|
response_data = self.model_monitoring_repository.transform(
|
|
model_name, data, model_retention)
|
|
|
|
self.logger.debug(response_data)
|
|
|
|
return response_data
|
|
|
|
@activity.defn(name="request_predict")
|
|
async def request_predict(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
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.
|
|
Returns:
|
|
dict[str, Any]: The predicted data.
|
|
"""
|
|
self.logger.info('Predicting data...')
|
|
data = DataFrame(input_data['data'])
|
|
model_name = input_data['model_name']
|
|
model_retention = input_data['model_retention']
|
|
|
|
self.logger.debug(data)
|
|
|
|
data.replace(np.nan, None, inplace=True)
|
|
|
|
response_data = self.model_monitoring_repository.predict(
|
|
model_name, data, model_retention)
|
|
|
|
self.logger.debug(response_data)
|
|
|
|
return response_data
|