SIENTIAPDE-1445
Update GITHUB_BRANCH to feature/SIENTIAPDE-1445 and enhance formatters with a new registry for workflow types, improving configuration management. Refactor process_schedules to utilize the new registry and add error handling for unsupported workflow types. Update tests to validate new functionality and ensure proper integration.
This commit is contained in:
vitor-aignosi
@@ -9,15 +9,22 @@ def common_config(config: dict[str, Any]):
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch')
|
||||
- schedule_name (str): Name of the schedule
|
||||
- frequency (str, optional): Frequency of execution (default: '1m')
|
||||
- max_retry_policy (int, optional): Maximum retry attempts (default: 1)
|
||||
- model_id (str): ID of the model
|
||||
- models (dict): Model configuration containing 'name' field
|
||||
- workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift')
|
||||
- schedule_name (str): Unique name identifier for this schedule
|
||||
- model_id (str): MongoDB ID of the associated model
|
||||
- model (dict): Model configuration containing:
|
||||
- name (str): Human-readable name of the model
|
||||
- model_config (dict, optional): Additional model-specific configuration
|
||||
- frequency (str, optional): Execution frequency (default: '1m')
|
||||
Format: '{number}{unit}' where unit is 's', 'm', 'h', or 'd'
|
||||
- offset (str, optional): Schedule offset/delay (default: '0m')
|
||||
- max_retry_policy (int, optional): Maximum retry attempts on failure (default: 1)
|
||||
- execution_timeout_seconds (int, optional): Workflow execution timeout (default: 300)
|
||||
- task_timeout_seconds (int, optional): Individual task timeout (default: 300)
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Common configuration dictionary with extracted parameters.
|
||||
dict[str, Any]: Common configuration dictionary with standardized parameters
|
||||
for Temporal workflow execution
|
||||
"""
|
||||
model = config['model']
|
||||
return {
|
||||
@@ -43,13 +50,14 @@ def drift(config: dict[str, Any]):
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- interval_minutes (int, optional): Detection interval in minutes (default: 60)
|
||||
- drift_metrics (list[str], optional): List of drift metrics to compute
|
||||
(default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
|
||||
- interval_minutes (int, optional): Time window for data comparison in minutes (default: 60)
|
||||
- drift_metrics (list[str], optional): Statistical metrics to compute (default:
|
||||
['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
|
||||
- Additional fields from common_config
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Drift configuration with workflow type set to 'drift'
|
||||
dict[str, Any]: Drift detection configuration with source/target tables,
|
||||
time interval, and metrics specifications. Results stored in 'drift_metrics' table
|
||||
"""
|
||||
return {
|
||||
**common_config(config),
|
||||
@@ -78,7 +86,9 @@ def simple_metrics(config: dict[str, Any]):
|
||||
- Additional fields from common_config
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Simple metrics configuration with workflow type set to 'simple_metrics'
|
||||
dict[str, Any]: Simple metrics configuration with source tables (predictions and
|
||||
actual data), target table, time interval, and metrics list. Results stored in
|
||||
'simple_metrics' table
|
||||
"""
|
||||
return {
|
||||
**common_config(config),
|
||||
@@ -97,12 +107,15 @@ def minimal_retrain(config: dict[str, Any]):
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- schedule_name (str): Name of the schedule
|
||||
- query (str): SQL query for retraining
|
||||
- query (str): SQL query to retrieve training data. Should return features
|
||||
and target variable in expected format
|
||||
- datetime_columns (list[str], optional): Column names to parse as datetime
|
||||
for proper temporal handling (default: [])
|
||||
- Additional fields from common_config
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Minimal retrain configuration with workflow type set to 'minimal_retrain'.
|
||||
dict[str, Any]: Minimal retrain configuration with SQL query, database settings,
|
||||
and datetime column specifications. Retraining logs are stored in 'log_retrain' table
|
||||
"""
|
||||
return {
|
||||
**common_config(config),
|
||||
@@ -112,9 +125,28 @@ def minimal_retrain(config: dict[str, Any]):
|
||||
'datetime_columns': config.get('datetime_columns', []),
|
||||
}
|
||||
|
||||
|
||||
def base_scouter(config: dict[str, Any]):
|
||||
"""
|
||||
Base scouter configuration.
|
||||
Build base scouter configuration shared by all scouter workflow types.
|
||||
|
||||
Creates the foundational configuration for OPC data collection workflows,
|
||||
including filter policies, database settings, and data retention parameters.
|
||||
This configuration is extended by specific scouter implementations (OPC UA, PI Web API).
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- filters (list[dict], optional): List of filter configurations with:
|
||||
- filter_name (str): Name of the filter
|
||||
- policy (str): Filter policy to apply
|
||||
- tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
|
||||
- debug_data_package (bool, optional): Enable debug data package logging (default: False)
|
||||
- fill_missing_tags (bool, optional): Fill missing tags with interpolation (default: False)
|
||||
- Additional fields from common_config
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Base scouter configuration with filters, database settings,
|
||||
and retention policies
|
||||
"""
|
||||
|
||||
filters = {}
|
||||
@@ -139,22 +171,19 @@ def scouter(config: dict[str, Any]):
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- filters (list[dict], optional): List of filter configurations
|
||||
- schedule_name (str): Name of the schedule (used for topic generation)
|
||||
- read_tags (list[dict]): List of tag configurations with:
|
||||
- filter_name (str): Name of the filter
|
||||
- policy (str): Filter policy
|
||||
- tag_name (str): Name of the tag
|
||||
- aggr_func (str, optional): Aggregation function (default: 'lts')
|
||||
- data_range (list[int], optional): Data range limits (default: [-100, 100])
|
||||
- tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
|
||||
- debug_data_package (bool, optional): Enable debug data package (default: False)
|
||||
- Additional fields from common_config
|
||||
- tag_name (str): Name of the tag to read
|
||||
- aggr_func (str, optional): Aggregation function for data collection (default: 'lts')
|
||||
Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum'
|
||||
- data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100])
|
||||
- Additional fields from base_scouter
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings.
|
||||
dict[str, Any]: OPC UA scouter configuration with Kafka topic, tag mappings,
|
||||
filters, and retention settings. Topic name follows pattern: 'raw_{schedule_name}'
|
||||
"""
|
||||
|
||||
|
||||
tags = {}
|
||||
for tag in config['read_tags']:
|
||||
tags[tag['tag_name']] = {
|
||||
@@ -168,9 +197,32 @@ def scouter(config: dict[str, Any]):
|
||||
'model_tags': tags,
|
||||
}
|
||||
|
||||
|
||||
def pi_web_api_scouter(config: dict[str, Any]):
|
||||
"""
|
||||
Build PI Web API scouter configuration from pipeline config.
|
||||
|
||||
Creates a data collection workflow configuration for OSIsoft PI servers using
|
||||
the PI Web API REST interface. Configures tag mappings with WebIDs, aggregation
|
||||
functions, and API query parameters including timeout management.
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- read_tags (list[dict]): List of tag configurations with:
|
||||
- tag_name (str): Name of the tag
|
||||
- webid (str): PI Web API WebID for the tag
|
||||
- aggr_func (str, optional): Aggregation function (default: 'lts')
|
||||
- data_range (list[int], optional): Valid data range (default: [-100, 100])
|
||||
- pi_web_api_config (dict): PI Web API connection settings with:
|
||||
- endpoint (str): PI Web API endpoint URL
|
||||
- period (str, optional): Time period for data retrieval (default: '*-1d')
|
||||
- max_count (int, optional): Maximum number of values to retrieve (default: 1)
|
||||
- api_timeout (int, optional): API request timeout in seconds
|
||||
- Additional fields from base_scouter
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
|
||||
API timeout is automatically adjusted to not exceed workflow frequency.
|
||||
"""
|
||||
tags = {}
|
||||
for tag in config['read_tags']:
|
||||
@@ -198,7 +250,7 @@ def pi_web_api_scouter(config: dict[str, Any]):
|
||||
'period': pi_web_api_config.get('period', '*-1d'),
|
||||
'max_count': pi_web_api_config.get('max_count', 1),
|
||||
'api_timeout': config_timeout,
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -227,13 +279,25 @@ def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[
|
||||
|
||||
def process_path_priority(path_priority: list[str]):
|
||||
"""
|
||||
Process and normalize path priority list to ensure it contains the required priorities.
|
||||
Process and normalize path priority list for filter policy execution order.
|
||||
|
||||
Validates and normalizes the priority list used to determine the order in which
|
||||
filter policies are evaluated in prediction workflows. Invalid priorities are
|
||||
removed, missing required priorities are appended, and the list is truncated to
|
||||
exactly 3 elements.
|
||||
|
||||
Valid priorities define workflow behavior when filters are triggered:
|
||||
- STOP: Halt workflow execution immediately
|
||||
- CONTINUE: Proceed to next step despite filter trigger
|
||||
- REPEAT: Retry the current step
|
||||
|
||||
Args:
|
||||
path_priority (list[str]): List of path priorities to process.
|
||||
path_priority (list[str]): User-provided list of path priorities. May contain
|
||||
invalid values or be incomplete.
|
||||
|
||||
Returns:
|
||||
list[str]: Normalized path priority list with exactly 3 elements: ["STOP", "CONTINUE", "REPEAT"].
|
||||
list[str]: Normalized path priority list with exactly 3 elements in user-specified
|
||||
or default order. Default order when priorities are missing: ["STOP", "CONTINUE", "REPEAT"]
|
||||
"""
|
||||
for priority in path_priority[:]:
|
||||
if priority not in ['STOP', 'CONTINUE', 'REPEAT']:
|
||||
@@ -252,20 +316,25 @@ def predictions_batch(config: dict[str, Any]):
|
||||
|
||||
Args:
|
||||
config (dict[str, Any]): Pipeline configuration containing:
|
||||
- write_tags (list[dict]): List of tag configurations with:
|
||||
- server_id (str): ID of the OPC server
|
||||
- type (str): Tag type ('prediction' or 'confidence')
|
||||
- addr (str): Tag address
|
||||
- data_type (str, optional): Data type (default: 'float')
|
||||
- path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"])
|
||||
- input_filters (list[dict], optional): List of input filter configurations
|
||||
- mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations
|
||||
- mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations
|
||||
- model_retention_minutes (int, optional): Model retention time in minutes (default: 60)
|
||||
- query (str): SQL query to retrieve input data for predictions
|
||||
- write_tags (list[dict]): List of OPC tag configurations for write-back with:
|
||||
- server_id (str): ID of the target OPC server
|
||||
- type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence)
|
||||
- addr (str): OPC tag address/path
|
||||
- data_type (str, optional): OPC data type (default: 'float')
|
||||
- datetime_columns (list[str], optional): Column names to parse as datetime (default: [])
|
||||
- path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"])
|
||||
- input_filters (list[dict], optional): Input data validation filters
|
||||
- mlflow_transform_filters (list[dict], optional): Transform stage filters
|
||||
- mlflow_predict_filters (list[dict], optional): Prediction stage filters
|
||||
- model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
|
||||
- save_transform (bool, optional): Save transformed data to database (default: True)
|
||||
- predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
|
||||
- Additional fields from common_config
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Predictions batch configuration with OPC output config, filters, and path priority.
|
||||
dict[str, Any]: Complete predictions batch configuration with OPC output mappings,
|
||||
multi-stage filters, SQL query, and retention policies
|
||||
"""
|
||||
tags: dict[str, Any] = {}
|
||||
for tag in config.get('write_tags', []):
|
||||
|
||||
Reference in New Issue
Block a user