gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #2 from Aignosi/SIENTIAPDE-1081-desenvolver-a-documentacao-do-modulo

Browse Source 
SIENTIAPDE-1081 Enhance documentation across multiple modules with detailed parameter descriptions and usage examples
This commit is contained in:
Matheus Demoner
2025-05-26 17:07:54 -03:00
committed by GitHub
parent bc3f26c280 49fa60c66f
commit dcb3e651ca
15 changed files with 358 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
README.md
View File

@@ -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
```

58
laborious/activities/gates.py
View File

@@ -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.
"""

16
laborious/activities/mlflow.py
View File

@@ -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.
"""
@@ -62,10 +62,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.
"""

22
laborious/activities/opc.py
View File

@@ -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'])

21
laborious/activities/postgres.py
View File

@@ -6,7 +6,6 @@ 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
@@ -49,10 +48,10 @@ class Postgres(BaseActivity):
Loads data from a custom query.
Args:
query (str): The query to load data from.
- query (str): The query to load data from.
Returns:
dict[str, dict]: The data from the query.
dict[str, Any]: The data from the query.
"""
self.logger.info(f"Fetching data from query: {query}")
@@ -95,10 +94,10 @@ class Postgres(BaseActivity):
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.
- 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
@@ -144,10 +143,10 @@ class Postgres(BaseActivity):
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.
- 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(

4
laborious/utils/connectors_config.py
View File

@@ -1,3 +1,7 @@
"""
Builds the configuration for the connectors.
"""
from os import getenv
import json

14
laborious/utils/filters/conditional_filters.py
View File

@@ -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

20
laborious/utils/filters/mlflow_filters.py
View File

@@ -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)

8
laborious/utils/logger.py
View File

@@ -4,6 +4,14 @@ import sys
def get_logger(name: str):
"""
Builds the logger. Gets the log level from the environment variable LOG_LEVEL.
If the log level is not set, it defaults to INFO.
Creates a stream handler and sets the log level.
Sets the formatter for the stream handler.
Adds the stream handler to the logger.
Returns the logger.
"""
log_level = getenv('LOG_LEVEL', 'INFO').upper()
logger = logging.getLogger(name)

17
laborious/utils/policies.py
View File

@@ -1,3 +1,20 @@
"""
Laborious Temporal Workflows Retry Policies
This module defines retry policies for Temporal workflows in the Laborious system.
These policies ensure reliable execution of ML prediction workflows by automatically
retrying failed operations with exponential backoff.
The retry_policy variable configures:
- Initial retry interval of 1 second
- Exponential backoff coefficient of 2.0
- Maximum retry interval capped at 1 minute
- Maximum of 1 retry attempt
This configuration helps prevent cascading failures while maintaining system responsiveness.
"""
from datetime import timedelta
from temporalio.common import RetryPolicy

78
laborious/utils/repository/model_repository.py
View File

@@ -1,9 +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.
@@ -13,7 +15,6 @@ import traceback
import mlflow
import pandas as pd
from sientia.ModelServing import ModelServing
from pathlib import Path
class MLFlowRepository():
@@ -49,7 +50,8 @@ class MLFlowRepository():
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.
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:
@@ -77,7 +79,9 @@ class MLFlowRepository():
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)
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
@@ -112,25 +116,26 @@ class MLFlowRepository():
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.
- 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.
- 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'.
- 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.
- 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
@@ -218,13 +223,16 @@ class MLFlowRepository():
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.
# 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'
# 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
@@ -257,10 +265,23 @@ class MLFlowRepository():
return metadata
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,
'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:
@@ -273,6 +294,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(

53
laborious/utils/repository/opc_repository.py
View File

@@ -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:

25
laborious/workflows/predictions_batch.py
View File

@@ -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

22
laborious/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -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.

26
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -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