diff --git a/.env.example b/.env.example index 8bd75e5..652ce20 100644 --- a/.env.example +++ b/.env.example @@ -11,9 +11,6 @@ MLFLOW_PORT="80" MLFLOW_USERNAME="aignosi" MLFLOW_PASSWORD="mlflow_password" -OPC_ID="1" -OPC_URL="opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840" - LOG_LEVEL="DEBUG" HTTP_METRICS_PORT="9090" HTTP_SDK_METRICS_PORT="9091" diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 3eaa27c..997202a 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -193,7 +193,31 @@ jobs: run: | python -m pip install --upgrade pip pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }} - pip install pytest pytest-cov pytest-asyncio + pip install -r requirements-dev.txt + + - name: 📝 Code Formatting Check (Ruff) + run: | + echo "Checking code formatting..." + ruff format --check model_manager/ tests/ + continue-on-error: false + + - name: 🔎 Code Linting (Ruff) + run: | + echo "Running linting checks..." + ruff check model_manager/ tests/ + continue-on-error: false + + - name: 🏷️ Type Checking (mypy) + run: | + echo "Running type checks..." + mypy model_manager/ + continue-on-error: true + + - name: 🔒 Security Analysis (Bandit) + run: | + echo "Running security analysis..." + bandit -r model_manager/ -ll -q + continue-on-error: true - name: 🧪 Run Tests with Pytest run: | diff --git a/README.md b/README.md index 61a50b9..28a38c3 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ A comprehensive AI model management platform for the complete machine learning l - **Temporal Workflow Orchestration**: Robust workflow management with automatic retry policies and fault tolerance - **Data Quality Gates**: Configurable filtering for data validation, MLFlow API responses, and custom validation rules - **Multi-Model Support**: Flexible ML model management with retention policies and versioning -- **Real-time Data Export**: PostgreSQL persistence and OPC server integration for industrial systems +- **Real-time Data Export**: PostgreSQL persistence for data storage - **Comprehensive Monitoring**: Prometheus metrics and detailed logging for operational visibility ### Advanced Capabilities @@ -19,6 +19,12 @@ A comprehensive AI model management platform for the complete machine learning l - **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support - **Model Retraining**: Automated model retraining workflows with production model updates +### Development & Quality Assurance +- **Code Quality Tools**: Ruff (linting/formatting), mypy (type checking), Bandit (security analysis) +- **Automated Validation**: Pre-commit validation script and CI/CD integration +- **Comprehensive Testing**: pytest with async support and 70%+ code coverage +- **Type Safety**: Static type checking with mypy for improved code reliability + ## Architecture The Model Manager system uses a Temporal-based workflow architecture with clear separation of concerns and robust error handling. The architecture is designed for high availability, scalability, and operational excellence in production ML environments. @@ -60,7 +66,6 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Task queue configuration and load balancing - Prometheus metrics server initialization - Notification handler setup and configuration - - OPC server connection management - **Key Features**: - Automatic scaling with `PollerBehaviorAutoscaling` - Health check endpoints for Kubernetes liveness/readiness probes @@ -83,20 +88,16 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - **Activities**: Main activity orchestrator combining all functionality through multiple inheritance - **Gates**: Data quality validation and filtering mechanisms - **MLFlow**: Model transformation and prediction operations -- **OPC**: Real-time data export to industrial OPC servers - **Key Features**: - Multiple inheritance pattern for unified activity interface - Configurable filter policies and validation rules - MLFlow model serving integration with configurable flavors - - OPC UA client with certificate-based authentication - Comprehensive error handling and notification integration - - Support for multiple OPC servers with independent configurations #### **Data Services (`model_manager/utils/`)** - **Connectors Config**: Environment variable-based configuration management -- **Repository**: Data access layer for MLFlow and OPC operations +- **Repository**: Data access layer for MLFlow operations - `model_repository.py`: MLFlow model operations and retraining - - `opc_repository.py`: OPC server communication and data writing - **Filters**: Data quality validation and MLFlow response filtering - `conditional_filters.py`: Input data validation filters - `mlflow_filters.py`: MLFlow API response validation filters @@ -105,14 +106,14 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Connection pool management and optimization - Security credential management - Configuration validation and error handling - - Support for multiple OPC servers and MLFlow model flavors + - Support for MLFlow model flavors ### Data Flow Architecture #### **1. Batch Prediction Pipeline** ``` Input Data (PostgreSQL) → Data Quality Gates → MLFlow Transform → -MLFlow Prediction → Response Validation → Export (PostgreSQL + OPC) +MLFlow Prediction → Response Validation → Export (PostgreSQL) ``` #### **2. Model Retraining Pipeline** @@ -121,16 +122,10 @@ Training Data → Model Retraining → Quality Validation → Production Update → Notification & Monitoring ``` -#### **3. Real-time Export Pipeline** -``` -Prediction Results → Data Formatting → OPC Server Write → -Success/Failure Metrics → Notification System -``` - ### Security Architecture #### **Authentication & Authorization** -- **Certificate-based OPC Authentication**: Secure industrial communication + - **MLFlow API Authentication**: Username/password with secure transmission - **Database Connection Security**: Encrypted connections with credential management - **Kubernetes Secrets Integration**: Secure credential storage and access @@ -195,11 +190,7 @@ The **PredictionsBatch** workflow is the main entry point for batch prediction p "API_ERROR": {"POLICY": "STOP"} }, "model_retention": 60, - "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "opc_output_config": { - "server_id": "opc_server_1", - "tags": ["prediction_output"] - } + "path_priority": ["STOP", "CONTINUE", "REPEAT"] } ``` @@ -267,8 +258,7 @@ The **PredictionProcess** workflow implements the core prediction pipeline for M "NAN_VALUES": {"POLICY": "STOP"} }, "model_retention": 60, - "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "opc_output_config": {...} + "path_priority": ["STOP", "CONTINUE", "REPEAT"] } ``` @@ -288,33 +278,30 @@ flowchart LR The **FormatAndExportPrediction** workflow handles prediction data formatting and export operations to multiple destinations. #### Purpose -- **Data Formatting**: Formats prediction data for different output destinations +- **Data Formatting**: Formats prediction data for database storage - **PostgreSQL Export**: Persists predictions to database with metrics -- **OPC Integration**: Writes predictions to OPC servers for real-time access - **Metrics Recording**: Tracks export operations and performance metrics #### Execution Flow 1. **Path Decision**: Determines formatting path based on configuration 2. **Data Formatting**: Formats prediction data for specific output requirements 3. **PostgreSQL Export**: Writes formatted predictions to database -4. **OPC Export**: Writes predictions to OPC servers -5. **Metrics Recording**: Records export performance and success metrics +4. **Metrics Recording**: Records export performance and success metrics #### Key Features - **Flexible Formatting**: Configurable output formats for different destinations -- **Multi-Destination Export**: PostgreSQL and OPC server integration +- **Database Export**: PostgreSQL integration for data persistence - **Performance Monitoring**: Comprehensive metrics for export operations - **Error Handling**: Robust error handling with notification integration #### Architecture Diagram ```mermaid flowchart LR - A[1. format_prediction/format_default_prediction] --> B[2. write_opc_data] --> C[3. export_data_to_postgres] --> D[4. write_metrics] + A[1. format_prediction/format_default_prediction] --> B[2. export_data_to_postgres] --> C[3. write_metrics] A -.-> Format[Data Formatting] - B -.-> OPC[OPC Servers] - C -.-> PostgreSQL[(PostgreSQL)] - D -.-> Prometheus[Prometheus] + B -.-> PostgreSQL[(PostgreSQL)] + C -.-> Prometheus[Prometheus] ``` ### 4. Minimal Retrain Workflow (`minimal_retrain.py`) @@ -351,7 +338,6 @@ flowchart LR - Temporal server/cluster - PostgreSQL database - MLFlow server -- OPC server(s) - MongoDB server (for notifications) **Note**: External dependencies must be available either through: @@ -398,12 +384,10 @@ flowchart LR 4. **Install Python dependencies** ```bash python -m pip install --upgrade pip + # Install production dependencies pip install -r requirements.txt - ``` - - 5. **Install test libraries** - ```bash - pip install pytest pytest-cov pytest-asyncio + # Install development and testing tools + pip install -r requirements-dev.txt ``` 4. **Create environment configuration file** @@ -499,6 +483,113 @@ fi python -m model_manager.worker.worker ``` +## 🔍 Code Quality & Validation + +### Overview + +Como Python não é uma linguagem compilada, utilizamos um conjunto robusto de ferramentas para validar a qualidade, segurança e correção do código antes da execução. Estas ferramentas detectam erros, problemas de estilo, vulnerabilidades de segurança e garantem a consistência do código. + +### Ferramentas de Validação + +#### 1. **Ruff** - Linting e Formatação ⚡ +Ferramenta moderna e extremamente rápida (escrita em Rust) que substitui múltiplas ferramentas: +- **Linting**: Detecta erros de código, problemas de estilo (PEP 8), bugs comuns +- **Formatação**: Formata código automaticamente de forma consistente +- **Velocidade**: 10-100x mais rápido que Flake8/Black + +#### 2. **mypy** - Type Checking 🏷️ +Verificador de tipos estáticos que analisa type hints: +- Detecta erros de tipo antes da execução +- Melhora a documentação do código +- Previne bugs relacionados a tipos incorretos + +#### 3. **Bandit** - Análise de Segurança 🔒 +Scanner de vulnerabilidades de segurança: +- Detecta padrões inseguros de código +- Identifica hardcoded passwords, SQL injection, etc. +- Garante conformidade com práticas de segurança + +#### 4. **pytest** - Testes Automatizados 🧪 +Framework de testes com cobertura de código: +- Executa testes unitários e de integração +- Mede cobertura de código +- Suporta testes assíncronos + +### Instalação das Ferramentas + +```bash +# Instalar dependências de desenvolvimento +pip install -r requirements-dev.txt +``` + +### Validação Completa + +#### Opção 1: Script Automatizado (Recomendado) +```bash +# Executar todas as validações de uma vez +./validate.sh +``` + +O script `validate.sh` executa automaticamente: +1. ✅ Verificação de formatação (Ruff) +2. ✅ Linting de código (Ruff) +3. ✅ Type checking (mypy) +4. ✅ Análise de segurança (Bandit) +5. ✅ Testes unitários com cobertura (pytest) + +#### Opção 2: Comandos Individuais +```bash +# 1. Verificar formatação +ruff format --check model_manager/ tests/ + +# 2. Verificar linting +ruff check model_manager/ tests/ + +# 3. Verificar tipos +mypy model_manager/ + +# 4. Análise de segurança +bandit -r model_manager/ -ll + +# 5. Executar testes +pytest tests/ --cov=model_manager --cov-report=term-missing +``` + +### Correção Automática + +Algumas ferramentas podem corrigir problemas automaticamente: + +```bash +# Formatar código automaticamente +ruff format model_manager/ tests/ + +# Corrigir problemas de linting automaticamente +ruff check --fix model_manager/ tests/ +``` + +### Configuração + +Todas as ferramentas são configuradas no arquivo `pyproject.toml`: +- **Ruff**: Regras de linting, formatação, complexidade +- **mypy**: Configurações de type checking +- **pytest**: Opções de teste e cobertura +- **Bandit**: Regras de segurança + +### Integração com CI/CD + +O workflow `.github/workflows/quality-gate.yml` executa automaticamente todas as validações em cada push/PR: +- ✅ Formatação e linting bloqueiam merge se falharem +- ⚠️ Type checking e segurança geram avisos mas não bloqueiam +- ✅ Testes devem passar com cobertura mínima de 70% + +### Boas Práticas + +1. **Antes de Commit**: Execute `./validate.sh` para garantir qualidade +2. **Durante Desenvolvimento**: Use `ruff check --watch` para feedback em tempo real +3. **Type Hints**: Adicione type hints em funções novas para melhor validação +4. **Testes**: Mantenha cobertura acima de 70% +5. **Segurança**: Revise e corrija todos os avisos do Bandit + ## 🧪 Testing ### Test Structure @@ -540,13 +631,6 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa - Labels: `pod_id`, `model_name`, `pipeline_name` - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] -### OPC Export Metrics -- `model_manager_prediction_opc_writing_count`: Counter for OPC server write operations - - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` -- `model_manager_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times - - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` - - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] - ### Data Quality Metrics - Filter pass/fail rates through notification system - MLFlow API response validation metrics @@ -571,14 +655,6 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa | `MLFLOW_PORT` | MLFlow server port | `5080` | Yes | | `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes | | `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes | -| `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No | -| `OPC_ID` | OPC server identifier | `1` | No | -| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No | -| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No | -| `OPC_CERT_PATH` | OPC client certificate path | `None` | No | -| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | -| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | -| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | | `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes | | `MONGODB_USERNAME` | MongoDB username | `root` | Yes | | `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | @@ -590,45 +666,6 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | | `POD_ID` | Kubernetes pod identifier | `None` | No | - - - -### OPC Configuration - -For multiple OPC servers, use the `OPC_CONFIG` environment variable: - -```json -{ - "opc_server_1": { - "url": "opc.tcp://server1:4840", - "name": "Server1", - "server_uri": "urn:server1:opcua", - "cert_path": "/path/to/cert.pem", - "private_key_path": "/path/to/key.pem", - "server_cert_path": "/path/to/server_cert.pem", - "reconnection_interval": 5000 - }, - "opc_server_2": { - "url": "opc.tcp://server2:4840", - "name": "Server2", - "server_uri": "urn:server2:opcua", - "cert_path": "/path/to/cert.pem", - "private_key_path": "/path/to/key.pem", - "server_cert_path": "/path/to/server_cert.pem", - "reconnection_interval": 5000 - } -} -``` - -For single OPC server, use individual environment variables: -- `OPC_URL` -- `OPC_NAME` -- `OPC_SERVER_URI` -- `OPC_CERT_PATH` -- `OPC_PRIVATE_KEY_PATH` -- `OPC_SERVER_CERT_PATH` -- `OPC_RECONNECTION_INTERVAL` - ### Workflow Configuration MongoDB pipeline configuration: @@ -705,7 +742,6 @@ This is the configuration created by the Orchestrator in Temporal. }, "model_id":"352", "model_name":"courier", - "opc_output_config":{}, "path_priority":["STOP","CONTINUE","REPEAT"], "predictions_storage_policy":"lts:1", "query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;", @@ -726,8 +762,7 @@ model_manager/ ├── activities/ # Temporal activity implementations │ ├── activities.py # Main activities orchestrator │ ├── gates.py # Data quality gates and filtering -│ ├── mlflow.py # MLFlow model operations -│ └── opc.py # OPC server operations +│ └── mlflow.py # MLFlow model operations ├── workflows/ # Temporal workflow definitions │ ├── predictions_batch.py # Main batch prediction workflow │ ├── minimal_retrain.py # Model retraining workflow @@ -742,8 +777,7 @@ model_manager/ │ │ ├── conditional_filters.py # Conditional data filters │ │ └── mlflow_filters.py # MLFlow response filters │ └── repository/ # Data access layer -│ ├── model_repository.py # MLFlow model operations -│ └── opc_repository.py # OPC server operations +│ └── model_repository.py # MLFlow model operations ├── metrics.py # Prometheus metrics definitions └── __init__.py ``` @@ -775,12 +809,7 @@ model_manager/ - Check connection credentials and network access - Ensure proper connection pool configuration -4. **OPC Connection Failures** - - Verify OPC server is accessible - - Check certificate and key file paths - - Review OPC server logs for connection issues - -5. **Workflow Execution Failures** +4. **Workflow Execution Failures** - Review activity error logs and notifications - Check data quality filter configurations - Verify input data format and required fields diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 6a8656d..81407a7 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -6,28 +6,25 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.logger import Logger from model_manager.activities.mlflow import MLFlow from model_manager.activities.gates import Gates - from model_manager.activities.opc import OPC from typing import Any -class Activities(Postgres, MLFlow, Gates, OPC): +class Activities(Postgres, MLFlow, Gates): """ Main activities orchestrator for the Model Manager system. This class combines functionality from multiple activity classes to provide a unified interface for all workflow operations. It manages database connections, - MLFlow model interactions, data quality validation, and OPC server communications. + MLFlow model interactions, and data quality validation. The class implements multiple inheritance to combine specialized functionality: - Postgres: Database operations and data persistence - MLFlow: Model inference and transformation operations - Gates: Data quality validation and filtering mechanisms - - OPC: Real-time data export to OPC servers Attributes: postgres_config (dict): PostgreSQL connection configuration mlflow_config (dict): MLFlow server configuration - opc_config (dict): OPC server configuration logger (Logger): Logging and observability instance notification_handler (NotificationHandler): Notification management instance """ @@ -35,7 +32,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): def __init__(self, postgres_config: dict[str, Any], mlflow_config: dict[str, Any], - opc_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler): """ @@ -49,8 +45,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): Required keys: host, port, user, password, dbname, min_connections, max_connections mlflow_config: MLFlow server configuration dictionary Required keys: host, port, username, password - opc_config: OPC server configuration dictionary - Can contain multiple server configurations logger: Logger instance for observability and debugging notification_handler: Notification handler for alerts and monitoring @@ -78,22 +72,15 @@ class Activities(Postgres, MLFlow, Gates, OPC): Gates.__init__(self, logger=logger, notification_handler=notification_handler) - OPC.__init__(self, - opc_servers=opc_config, - logger=logger, - notification_handler=notification_handler) - async def shutdown(self): """ Gracefully shutdown all activities and clean up resources. This method ensures proper cleanup of all resources including: - PostgreSQL connection pools - - OPC server connections - Any other resources that need explicit cleanup The method should be called before the application terminates to ensure proper resource cleanup and prevent resource leaks. """ Postgres.close(self) - await OPC.shutdown(self) diff --git a/model_manager/activities/opc.py b/model_manager/activities/opc.py deleted file mode 100644 index 2222eed..0000000 --- a/model_manager/activities/opc.py +++ /dev/null @@ -1,356 +0,0 @@ -from temporalio import activity, workflow - - -with workflow.unsafe.imports_passed_through(): - from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler - from sientia_do.notifications.models import NotificationLevel - from sientia_do.temporal.activities.base import BaseActivity - from sientia_do.observability.logger import Logger - from model_manager.utils.repository.opc_repository import OpcRepository - from typing import Any - import traceback - from pandas import DataFrame - -OPC_WRITTING_ERROR_CONFIDENCE = 12 - - -class OPC(BaseActivity): - """ - OPC server integration activities for real-time data export. - - This class provides comprehensive OPC UA client functionality for connecting - to multiple OPC servers and writing prediction data in real-time. It implements - secure communication with certificate-based authentication and automatic - reconnection capabilities. - - The class supports multiple OPC servers with individual configurations and - provides robust error handling and monitoring for production environments. - - Attributes: - opc_servers (dict): Configuration for multiple OPC servers - opc_repository (dict): Active OPC repository connections - logger (Logger): Logging and observability instance - notification_handler (NotificationHandler): Notification management instance - """ - - def __init__(self, opc_servers: dict[str, dict[str, Any]], - logger: Logger, notification_handler: NotificationHandler): - - self.logger = logger - self.notification_handler = notification_handler - self.opc_servers = opc_servers - - BaseActivity.__init__( - self, logger, notification_handler, set_error_counter=True) - - self.opc_repository: dict[str, OpcRepository] = {} - self.opc_servers = opc_servers - - async def init_opc(self): - """ - Initialize OPC server connections and establish communication channels. - - This method iterates through all configured OPC servers and attempts to - establish secure connections using certificate-based authentication. - Each server connection is managed independently, and connection failures - are reported through the notification system. - - The method performs the following operations: - 1. Creates OpcRepository instances for each configured server - 2. Establishes secure connections with certificate validation - 3. Reports connection success/failure through notifications - 4. Logs connection status for operational visibility - - Raises: - Exception: If OPC repository initialization fails or connection - establishment encounters critical errors - - Note: - Connection failures are logged and reported but do not prevent - the initialization of other OPC servers. Each server is handled - independently to ensure maximum availability. - """ - self.logger.info("Initializing OPC servers...") - for id, server in self.opc_servers.items(): - self.opc_repository[id] = OpcRepository( - id=server['id'], - url=server['url'], - logger=self.logger, - server_uri=server['server_uri'], - cert_path=server['cert_path'], - private_key_path=server['private_key_path'], - server_cert_path=server['server_cert_path'], - notification_handler=self.notification_handler, - reconnection_interval=server['reconnection_interval'], - pod_id=self.pod_id - ) - is_connected, error_data = await self.opc_repository[id].connect() - if not is_connected: - self.send_notification( - metadata={ - 'model_id': '-', - 'model_name': '-', - 'workflow_name': '-', - 'schedule_name': 'INITIALIZATION' - }, - notification_id=error_data['notification_id'], - message=error_data['message'], - block=error_data['block'], - level=error_data.get('level', NotificationLevel.ERROR), - attachment_content=error_data.get( - 'attachment_content', None) - ) - else: - self.logger.info( - f"OPC server {id} connected successfully.") - - async def write_data(self, server_id: str, tag: str, data: Any, - data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: - """ - Write data to a specific OPC server tag with comprehensive error handling. - - This method provides a secure and reliable way to write data to OPC servers - with automatic error handling, notification integration, and detailed logging. - It validates server availability before attempting write operations and - provides comprehensive error reporting for operational monitoring. - - Args: - - server_id (str): The id 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. - - Returns: - - bool: True if the data was written successfully, False otherwise. - """ - - try: - is_success, error_data = await self.opc_repository[server_id].write_data( - tag, data, data_type, self.logger, metadata) - if not is_success: - self.send_notification( - metadata=metadata, - notification_id=error_data['notification_id'], - message=error_data['message'], - block=error_data['block'], - level=error_data.get('level', NotificationLevel.ERROR), - attachment_content=error_data.get( - 'attachment_content', None) - ) - return False - return True - except Exception as e: - trace = traceback.format_exc() - self.send_notification( - metadata=metadata, - notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR", - message=f"Error writing data to OPC server: {e}", - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=trace - ) - raise e - - def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool: - """ - Validate that an OPC server is available and configured for write operations. - - This method checks if the specified OPC server exists in the active - repository and is available for data writing operations. It provides - immediate feedback for server availability and logs validation failures - for operational monitoring. - - Args: - server_id (str): Unique identifier for the OPC server to validate - metadata (dict[str, Any]): Context metadata for logging and notifications - - Returns: - bool: True if server is available, False otherwise - - Note: - Server validation failures are automatically reported through the - notification system with detailed information about available servers. - This helps operators quickly identify configuration issues. - """ - if self.opc_repository.get(server_id) is None: - message = f"OPC server {server_id} not found to perform write operation." - self.send_notification( - metadata=metadata, - notification_id="OPC_SERVER_NOT_FOUND", - message=message, - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=f"OPC servers: {list(self.opc_repository.keys())}" - ) - return False - return True - - async def manage_output_tags( - self, server_id: str, config: dict[str, Any], data: DataFrame, - metadata: dict[str, Any], success: bool) -> tuple[bool, int]: - """ - Manage the writing of prediction and confidence data to OPC server tags. - - This method orchestrates the writing of multiple data types to OPC servers - based on configuration. It handles both prediction data and confidence - values independently, allowing for flexible tag configuration and - comprehensive error handling. - - The method supports two main tag types: - 1. Prediction tags: Write actual prediction values to configured OPC tags - 2. Confidence tags: Write confidence scores to separate OPC tags - - Args: - server_id (str): Unique identifier for the target OPC server - config (dict[str, Any]): OPC tag configuration containing: - - prediction_tags (dict, optional): Prediction tag configurations - - confidence_tags (dict, optional): Confidence tag configurations - data (DataFrame): DataFrame containing prediction and confidence data - metadata (dict[str, Any]): Context metadata for logging and notifications - success (bool): Current success status to maintain across operations - - Returns: - tuple[bool, int]: (overall_success, total_tags_written) - - overall_success: True if all configured tags were written successfully - - total_tags_written: Count of successfully written tags - """ - - count = 0 - if 'prediction_tags' in config: - for tag, tag_config in config['prediction_tags'].items(): - local_success = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - metadata=metadata - ) - if local_success: - self.info( - f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) - count += 1 - success = success and local_success - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - local_success = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction_confidence'].values[0], - data_type=tag_config['data_type'], - tag_type='confidence', - metadata=metadata - ) - if local_success: - self.info( - f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) - count += 1 - success = success and local_success - - return success, count - - @activity.defn(name='write_opc_data') - async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: - """ - Write prediction and confidence data to OPC servers. The two writing - 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 - to the OPC servers. - - 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. - - Returns: - - dict[Any, Any]: The data that was written to the OPC servers. - - """ - metadata = input_data['metadata'] - self.info("Writing data to OPC servers...", metadata) - data = DataFrame(input_data['data']) - opc_output_config = input_data['opc_output_config'] - self.info(f"Data to write: {data.size} rows", metadata) - - success = True - - for server_id, config in opc_output_config.items(): - - if not self.validate_server(server_id, metadata): - success = False - continue - - local_success, local_count = await self.manage_output_tags( - server_id, config, data, metadata, success) - success = success and local_success - - self.info( - f"Process completed for OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata) - - return self.process_confidence(data, success, metadata) - - def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]: - """ - Process prediction confidence based on OPC write operation success. - - This method updates the prediction confidence values in the DataFrame - based on the success status of OPC server write operations. If any - write operations failed, it sets the confidence to a predefined error - value to indicate data quality issues. - - The method implements a confidence degradation strategy: - - Success: Maintains original confidence values - - Failure: Sets confidence to error value for operational awareness - - Args: - data (DataFrame): DataFrame containing prediction and confidence data - success (bool): Overall success status of OPC write operations - metadata (dict[str, Any]): Context metadata for logging and notifications - - Returns: - dict[Any, Any]: Processed data as a dictionary with updated confidence values - - Note: - The error confidence value (OPC_WRITTING_ERROR_CONFIDENCE = 12) is - used to indicate that data was not successfully exported to OPC servers. - This allows downstream systems to handle data quality appropriately. - """ - - if not success: - data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - self.debug( - f"Some data could not be written to OPC servers, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.", - metadata - ) - - else: - self.debug("Data written to OPC servers successfully.", metadata) - - return data.to_dict() - - async def shutdown(self): - """ - Gracefully shutdown all OPC server connections and cleanup resources. - - This method ensures proper cleanup of all active OPC server connections - by calling the disconnect method on each repository instance. It's - designed to be called during application shutdown to prevent resource - leaks and ensure clean termination. - - The method performs the following cleanup operations: - 1. Iterates through all active OPC repository connections - 2. Calls disconnect() on each repository instance - 3. Allows for graceful connection termination - 4. Prevents resource leaks and connection hanging - - Note: - This method should be called during application shutdown to ensure - proper cleanup. It handles all active connections regardless of - their current state and provides a clean shutdown experience. - """ - for opc in self.opc_repository.values(): - await opc.disconnect() diff --git a/model_manager/metrics.py b/model_manager/metrics.py index 96b806f..a19b03c 100644 --- a/model_manager/metrics.py +++ b/model_manager/metrics.py @@ -13,14 +13,13 @@ Key Metric Categories: - Application Health: Overall system status and availability - Prediction Operations: Count and performance of prediction operations - Data Quality: Confidence levels and validation results -- Export Operations: Database and OPC export performance +- Export Operations: Database export performance - Response Times: Performance monitoring for various operations Metric Labels: - pod_id: Kubernetes pod identifier for multi-instance deployments - model_name: Name of the ML model being used - pipeline_name: Name of the prediction pipeline -- opc_server_id: Identifier for OPC server operations """ from prometheus_client import Gauge, Counter, Histogram @@ -56,17 +55,3 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram( CORE_LABELS, buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] ) - -# OPC export metrics -PREDICTION_OPC_WRITING_COUNT = Counter( - "model_manager_prediction_opc_writing_count", - "Number of predictions written to the OPC server", - [*CORE_LABELS, "opc_server_id"], -) - -PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR = Histogram( - "model_manager_prediction_opc_writing_response_time_monitor", - "Current response time of each prediction written to the OPC server", - [*CORE_LABELS, "opc_server_id"], - buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] -) diff --git a/model_manager/utils/connectors_config.py b/model_manager/utils/connectors_config.py index 145f248..40e7c07 100644 --- a/model_manager/utils/connectors_config.py +++ b/model_manager/utils/connectors_config.py @@ -59,45 +59,6 @@ def build_mlflow_config() -> Dict[str, Any]: } -def build_opc_config() -> Dict[str, Any]: - """ - Build OPC server configuration from environment variables. - - This function constructs an OPC server configuration dictionary from - environment variables. It supports both single server and multi-server - configurations with flexible parameter handling. - - Environment Variables: - OPC_CONFIG: JSON string containing multiple OPC server configurations - OPC_ID: OPC server ID (fallback, default: 1) - OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840) - OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840) - OPC_CERT_PATH: Client certificate path (fallback, default: None) - OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None) - OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None) - OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120) - - Returns: - dict: OPC server configuration dictionary - """ - opc_raw = getenv('OPC_CONFIG', None) - - if opc_raw: - return json.loads(opc_raw) - - return { - getenv('OPC_ID', '1'): { - 'id': getenv('OPC_ID', '1'), - 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), - 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), - 'cert_path': getenv('OPC_CERT_PATH', None), - 'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None), - 'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None), - 'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')) - } - } - - def build_mongodb_config() -> Dict[str, Any]: """ Build MongoDB configuration from environment variables. diff --git a/model_manager/utils/repository/opc_repository.py b/model_manager/utils/repository/opc_repository.py deleted file mode 100644 index 94b6c46..0000000 --- a/model_manager/utils/repository/opc_repository.py +++ /dev/null @@ -1,359 +0,0 @@ -import asyncio -import traceback -import time -from datetime import datetime -from pathlib import Path -from typing import Any -from asyncua import Client -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from asyncua.ua import DataValue, Variant, VariantType, DateTime -from regex import F -from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler -from sientia_do.notifications.models import NotificationLevel -from sientia_do.observability.logger import Logger -from model_manager import metrics - -data_type_map = { - 'float': { - 'converter': float, - 'opc_type': VariantType.Float, - }, - 'double': { - 'converter': float, - 'opc_type': VariantType.Double, - }, - 'int': { - 'converter': int, - 'opc_type': VariantType.Int32, - }, - 'bool': { - 'converter': bool, - 'opc_type': VariantType.Boolean, - }, - 'str': { - 'converter': str, - 'opc_type': VariantType.String, - } -} - - -class OpcRepository(): - def __init__(self, id: 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, pod_id: str = None): - self.url = url - self.id = id - self.server_uri = server_uri - self.cert_path = cert_path - self.private_key_path = private_key_path - self.server_cert_path = server_cert_path - self.logger = logger - self.error_count = 0 - self.reconnection_interval = reconnection_interval - self.last_reconnection_time = None - self.notification_handler = notification_handler - self.client = None - self.pod_id = pod_id - - self.metadata = { - 'model_name': '-', - 'model_id': '-', - 'workflow_name': 'opc_repository', - 'schedule_name': '-' - } - - async def set_security(self): - """ - Configures the security settings for the OPC UA client. - This method sets up the security policy, certificates, and timeouts - required for establishing a secure connection with the OPC UA server. - 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. - Security Settings: - - Security Policy: Basic256 - - Secure Channel Timeout: 10,000,000 ms - - Session Timeout: 10,000,000 ms - """ - - if not all([self.cert_path, self.private_key_path]): - raise ValueError( - "Certificate and private key paths must be provided for secure connection.") - cert = Path(self.cert_path) - private_key = Path(self.private_key_path) - server_cert = Path( - self.server_cert_path) if self.server_cert_path else None - - self.client.application_uri = self.server_uri - self.logger.custom_info('Setting security...', self.metadata) - await self.client.set_security( - SecurityPolicyBasic256, - certificate=str(cert), - private_key=str(private_key), - server_certificate=str(server_cert) - ) - self.client.secure_channel_timeout = 10000000 - self.client.session_timeout = 10000000 - - async def connect(self) -> tuple[bool, dict[str, Any]]: - """ - Establishes a connection to the OPC server. - This method initializes the OPC client using the provided URL and - sets up security if a certificate path is specified. It then - attempts to connect to the server and logs the connection status. - Raises: - Exception: If the connection to the OPC server fails. - """ - - self.client = Client(self.url) - if self.cert_path: - await self.set_security() - self.logger.custom_info( - f'Starting connection to OPC server {self.id}...', self.metadata) - return await self.try_connect() - - async def try_connect(self) -> tuple[bool, dict[str, Any]]: - """ - Attempt to establish connection to the OPC server. - - This method performs the actual connection attempt to the OPC server - and handles connection failures with comprehensive error reporting. - It updates reconnection timing and provides detailed error information - for operational monitoring and debugging. - - Returns: - tuple[bool, dict[str, Any]]: Connection result - - bool: True if connection successful, False otherwise - - dict: Error information if connection failed - """ - - try: - self.last_reconnection_time = datetime.now() - await self.client.connect() - return True, {} - except Exception as e: - trace = traceback.format_exc() - self.logger.custom_error(trace, self.metadata) - - return False, { - "notification_id": f"OPC_CONNECTION_ERROR_{self.id}", - "message": f"Failed to connect to OPC server: {e}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - async def disconnect(self): - """ - Gracefully disconnect from the OPC server. - - This method safely terminates the connection to the OPC server - and cleans up client resources. It handles disconnection errors - gracefully and ensures proper resource cleanup. - """ - if self.client is None: - return - try: - await self.client.disconnect() - self.logger.custom_info( - 'Disconnected from OPC server', self.metadata) - except Exception as e: - self.logger.custom_error( - f"Failed to disconnect from OPC server: {e}", self.metadata) - self.client = None - - async def validate_connection(self) -> tuple[bool, dict[str, Any]]: - """ - Validate and maintain OPC server connection health. - - This method performs comprehensive connection validation and - implements automatic reconnection logic for production reliability. - It handles various connection states and implements intelligent - reconnection strategies with error counting and timing controls. - - Connection Validation: - 1. Checks client existence and connection state - 2. Implements error counting with automatic disconnection - 3. Enforces reconnection timing windows - 4. Provides detailed error reporting and notifications - - Reconnection Strategy: - - Error Count Threshold: Disconnects after 5 consecutive errors - - Reconnection Window: Enforces minimum intervals between attempts - - Automatic Recovery: Attempts reconnection when conditions allow - - State Monitoring: Continuously monitors connection health - - Args: - None - - Returns: - tuple[bool, dict[str, Any]]: Connection validation result - - bool: True if connection is healthy, False otherwise - - dict: Error information if validation fails - """ - if self.client is None: - return await self.connect() - - if self.error_count > 5: - self.logger.custom_warning( - f"OPC server {self.id} will be disconnected due to multiple errors", self.metadata) - try: - await self.disconnect() - except Exception as e: - trace = traceback.format_exc() - self.logger.custom_error( - f"Failed to disconnect from OPC server: {e}", self.metadata) - self.logger.custom_error(trace, self.metadata) - self.logger.custom_info( - f"Attempting to reconnect to OPC server {self.id}...", self.metadata) - return await self.connect() - - # Check if client is connected using asyncua's connection state - try: - if self.client.uaclient.protocol is None or self.client.uaclient.protocol.state == "closed": - # OPC server is not connected - self.logger.custom_error( - f"OPC server {self.id} is not connected", self.metadata) - if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( - ) > self.reconnection_interval: - await self.disconnect() - self.logger.custom_info( - f"Trying to reconnect to OPC server {self.id}...", self.metadata) - return await self.connect() - - return False, { - "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}", - "message": f"OPC server {self.id} is not connected, waiting for next reconnection window...", - "block": "opc_repository", - "level": NotificationLevel.WARNING - } - return True, {} - except Exception as e: - trace = traceback.format_exc() - message = f"Failed to validate connection to OPC server: {e}" - self.logger.custom_error(message, self.metadata) - return False, { - "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{self.id}", - "message": message, - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - async def write_data(self, node: str, value: Any, data_type: str, - logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]: - """ - Write data to OPC server with comprehensive validation and monitoring. - - This method provides secure and reliable data writing to OPC servers - with automatic connection validation, data type conversion, and - comprehensive error handling. It implements performance monitoring - and metrics collection for operational visibility. - - Data Writing Process: - 1. Connection validation and automatic reconnection - 2. Node validation and error handling - 3. Data type conversion and validation - 4. OPC data writing with timestamp - 5. Performance metrics collection - 6. Error handling and notification - - Args: - node (str): OPC node identifier to write data to - value (Any): Data value to write to the OPC node - data_type (str): Data type for OPC conversion - logger (Logger): Logger instance for operation logging - metadata (dict[str, Any]): Context metadata for logging and metrics - - Returns: - tuple[bool, dict[str, Any]]: Write operation result - - bool: True if write successful, False otherwise - - dict: Error information if write failed - """ - - is_connected, error = await self.validate_connection() - - if not is_connected: - return False, error - - start_time = time.time() - - try: - node_obj = self.client.get_node(node) - except Exception as e: - trace = traceback.format_exc() - logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) - self.error_count += 1 - return False, { - "notification_id": f"OPC_WRITE_GET_NODE_ERROR_{self.id}", - "message": f"Failed to get node from OPC server: {e} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - if data_type not in data_type_map: - return False, { - "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", - "message": f"Unsupported data type: {data_type} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR - } - - data = data_type_map[data_type]['converter'](value) - logger.custom_info( - f'Writing {data} - {type(data)} to {node}', metadata) - now = datetime.now() - ua_data = DataValue( - Variant(data, data_type_map[data_type]['opc_type']), - SourceTimestamp=DateTime( - now.year, - now.month, - now.day, - now.hour, - now.minute, - now.second, - now.microsecond - ) - ) - - try: - await node_obj.write_value(ua_data) - - metrics.PREDICTION_OPC_WRITING_COUNT.labels( - pod_id=self.pod_id, - model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], - opc_server_id=self.id - ).inc() - - end_time = time.time() - response_time = end_time - start_time - metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels( - pod_id=self.pod_id, - model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], - opc_server_id=self.id - ).observe(response_time) - - except Exception as e: - trace = traceback.format_exc() - logger.custom_error(trace, metadata) - self.error_count += 1 - return False, { - "notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}", - "message": f"Failed to write data to OPC server: {e} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - self.error_count = 0 - - return True, {} diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 5d1d0b1..92631ab 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -42,7 +42,6 @@ with workflow.unsafe.imports_passed_through(): from model_manager.utils.connectors_config import ( build_postgres_config, build_mlflow_config, - build_opc_config, build_mongodb_config ) from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler @@ -63,9 +62,8 @@ async def main(): 2. Starts Prometheus metrics server 3. Initializes notification handler 4. Creates and configures activities - 5. Initializes OPC connections - 6. Starts Temporal client and workers - 7. Manages worker lifecycle and graceful shutdown + 5. Starts Temporal client and workers + 6. Manages worker lifecycle and graceful shutdown The function runs indefinitely until interrupted or an error occurs. On error, it performs cleanup and exits with a non-zero status code. @@ -105,14 +103,10 @@ async def main(): activities = Activities( postgres_config=build_postgres_config(), mlflow_config=build_mlflow_config(), - opc_config=build_opc_config(), logger=logger, notification_handler=notification_handler ) - logger.custom_info('Initializing OPC...', metadata) - await activities.init_opc() - logger.custom_info( f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) @@ -167,8 +161,6 @@ async def main(): activities.format_prediction, activities.format_default_prediction, activities.get_last_timestamp, - # OPC - activities.write_opc_data, # Postgres activities.load_custom_query, activities.repeat_last_prediction, diff --git a/model_manager/workflows/predictions_batch.py b/model_manager/workflows/predictions_batch.py index 281b187..dec0fb7 100644 --- a/model_manager/workflows/predictions_batch.py +++ b/model_manager/workflows/predictions_batch.py @@ -58,7 +58,7 @@ class PredictionsBatch(): - mlflow_predict_filters (dict, optional): MLFlow prediction filters - model_retention (int, optional): Model retention period in minutes - path_priority (list[str]): Decision path priority configuration - - opc_output_config (dict, optional): OPC server export configuration + - datetime_columns (list[str], optional): Columns to treat as datetime Returns: @@ -115,7 +115,7 @@ class PredictionsBatch(): }), 'model_config': input_data.get('model_config', {}), 'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']), - 'opc_output_config': input_data.get('opc_output_config', {}), + 'prediction_store_policy': input_data.get( 'prediction_store_policy', 'lts:1') } diff --git a/model_manager/workflows/sub_workflows/format_and_export_prediction.py b/model_manager/workflows/sub_workflows/format_and_export_prediction.py index ef195c4..60fad76 100644 --- a/model_manager/workflows/sub_workflows/format_and_export_prediction.py +++ b/model_manager/workflows/sub_workflows/format_and_export_prediction.py @@ -14,9 +14,9 @@ class FormatAndExportPrediction(): Data formatting and export workflow for prediction results. This workflow handles the final stages of the prediction pipeline, including - data formatting, database persistence, OPC server export, and metrics recording. + data formatting, database persistence, and metrics recording. It implements flexible formatting based on prediction quality and provides - comprehensive export capabilities to multiple destinations. + comprehensive export capabilities. The workflow supports two main prediction paths: 1. Normal Prediction: Formats and exports successful prediction results @@ -24,7 +24,6 @@ class FormatAndExportPrediction(): Export Destinations: - PostgreSQL Database: Persistent storage with timestamp conversion - - OPC Servers: Real-time industrial system integration - Prometheus Metrics: Performance monitoring and operational visibility """ @@ -36,14 +35,12 @@ class FormatAndExportPrediction(): This method orchestrates the complete data export process by: 1. Determining the appropriate formatting strategy based on path_flag 2. Formatting prediction data according to quality and requirements - 3. Exporting data to OPC servers for real-time industrial access - 4. Persisting data to PostgreSQL database with comprehensive metadata - 5. Recording performance metrics for operational monitoring + 3. Persisting data to PostgreSQL database with comprehensive metadata + 4. Recording performance metrics for operational monitoring The method implements flexible formatting strategies: - Normal predictions: Full data formatting with confidence scores - Error predictions: Default formatting with error indicators - - Comprehensive export: Multi-destination data distribution Args: input_data: Complete configuration for the export workflow @@ -58,7 +55,6 @@ class FormatAndExportPrediction(): - comment (str): Operational comment or error description - schema (str): Database schema for data storage - table_name (str): Target table for data persistence - - opc_output_config (dict[str, Any]): OPC server export configuration - prediction_store_policy (str, optional): Data retention policy Returns: @@ -100,18 +96,6 @@ class FormatAndExportPrediction(): start_to_close_timeout=timedelta(seconds=60) ) - # write to opc - prediction = await workflow.execute_activity_method( - Activities.write_opc_data, - { - **metadata, - 'opc_output_config': input_data['opc_output_config'], - 'data': prediction - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60) - ) - # write to postgres await workflow.execute_activity_method( Activities.export_data_to_postgres, diff --git a/model_manager/workflows/sub_workflows/prediction_process.py b/model_manager/workflows/sub_workflows/prediction_process.py index e613ab5..c5407ec 100644 --- a/model_manager/workflows/sub_workflows/prediction_process.py +++ b/model_manager/workflows/sub_workflows/prediction_process.py @@ -64,7 +64,7 @@ class PredictionProcess(): - mlflow_predict_filters (dict): MLFlow prediction filters - model_retention (int): Model retention period in minutes - path_priority (list[str]): Decision path priority configuration - - opc_output_config (dict): OPC server export configuration + Returns: None: The workflow completes successfully when export workflow finishes @@ -210,7 +210,6 @@ class PredictionProcess(): 'model_id': model_id, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': input_data['opc_output_config'], 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'comment': comment, @@ -287,7 +286,6 @@ class PredictionProcess(): 'schema': schema, 'table_name': table_name, 'comment': comment, - 'opc_output_config': input_data['opc_output_config'], 'prediction_store_policy': input_data['prediction_store_policy'] } ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7b577c6 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,154 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "model-manager" +version = "0.0.0" +description = "Sientia DataOps Model Manager - ML Model Orchestration System" +readme = "README.md" +requires-python = ">=3.11" +authors = [ + {name = "Aignosi", email = "dev@aignosi.com"} +] + +[tool.ruff] +line-length = 100 +target-version = "py311" +exclude = [ + ".git", + ".venv", + "venv", + "__pycache__", + "*.pyc", + ".pytest_cache", + "htmlcov", +] + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "UP", # pyupgrade + "N", # pep8-naming + "YTT", # flake8-2020 + "S", # flake8-bandit + "BLE", # flake8-blind-except + "A", # flake8-builtins + "C90", # mccabe complexity +] + +ignore = [ + "E501", # line too long (handled by formatter) + "S101", # use of assert (needed for tests) + "S105", # possible hardcoded password (false positives) + "S106", # possible hardcoded password (false positives) + "N802", # function name should be lowercase (temporal decorators) + "N806", # variable in function should be lowercase +] + +[tool.ruff.lint.per-file-ignores] +"tests/**/*.py" = [ + "S101", # assert allowed in tests + "S105", # hardcoded passwords ok in tests + "S106", # hardcoded passwords ok in tests +] + +[tool.ruff.lint.mccabe] +max-complexity = 15 + +[tool.ruff.format] +quote-style = "single" +indent-style = "space" +line-ending = "auto" + +[tool.mypy] +python_version = "3.11" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = false +disallow_incomplete_defs = false +check_untyped_defs = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +strict_equality = true +ignore_missing_imports = false + +# Ignore missing imports for external packages +[[tool.mypy.overrides]] +module = "temporalio.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "sientia_do.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "mlflow.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "prometheus_client.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "redis.*" +ignore_missing_imports = true + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "-v", + "--strict-markers", + "--cov=model_manager", + "--cov-report=term-missing", + "--cov-report=html", + "--cov-report=xml", +] +markers = [ + "asyncio: marks tests as async", + "integration: marks tests as integration tests", + "unit: marks tests as unit tests", +] + +[tool.coverage.run] +source = ["model_manager"] +omit = [ + "*/tests/*", + "*/venv/*", + "*/__pycache__/*", + "*/site-packages/*", +] +branch = true + +[tool.coverage.report] +precision = 2 +show_missing = true +skip_covered = false +exclude_lines = [ + "pragma: no cover", + "def __repr__", + "def __str__", + "raise AssertionError", + "raise NotImplementedError", + "if __name__ == .__main__.:", + "if TYPE_CHECKING:", + "class .*\\bProtocol\\):", + "@(abc\\.)?abstractmethod", +] + +[tool.coverage.html] +directory = "htmlcov" + +[tool.bandit] +exclude_dirs = ["tests", "venv", ".venv"] +skips = ["B101", "B601"] # Skip assert and shell injection in controlled environments diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..e65402b --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,17 @@ +# Development and Testing Dependencies +# These packages are only needed for development, testing, and code quality checks +# Install with: pip install -r requirements-dev.txt + +# Code Quality & Linting +ruff>=0.1.0 # Fast Python linter and formatter (replaces flake8, black, isort) +mypy>=1.7.0 # Static type checker +bandit>=1.7.5 # Security vulnerability scanner + +# Testing +pytest>=7.4.0 # Testing framework +pytest-cov>=4.1.0 # Coverage plugin for pytest +pytest-asyncio>=0.21.0 # Async test support (already in main requirements) + +# Development Tools +ipython>=8.12.0 # Enhanced Python shell +ipdb>=0.13.13 # IPython debugger diff --git a/requirements.txt b/requirements.txt index 1ce9f9c..8fcde0c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,6 @@ temporalio psycopg2-binary sqlalchemy -asyncua redis git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index 106ccf7..b6e53df 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -4,14 +4,12 @@ from sientia_do.temporal.activities.postgres import Postgres from model_manager.activities.activities import Activities from model_manager.activities.mlflow import MLFlow from model_manager.activities.gates import Gates -from model_manager.activities.opc import OPC @patch('model_manager.activities.activities.Postgres.__init__') @patch('model_manager.activities.activities.MLFlow.__init__') -@patch('model_manager.activities.activities.OPC.__init__') @patch('model_manager.activities.activities.Gates.__init__') -def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgres_init): +def test___init__(mock_gates_init, mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', @@ -30,19 +28,12 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre '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 ) @@ -50,7 +41,6 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre assert isinstance(activities, Activities) assert isinstance(activities, Postgres) assert isinstance(activities, MLFlow) - assert isinstance(activities, OPC) assert isinstance(activities, Gates) mock_postgres_init.assert_called_once_with( @@ -76,13 +66,6 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre notification_handler=notification_handler ) - mock_opc_init.assert_called_once_with( - ANY, - opc_servers=opc_config, - logger=logger, - notification_handler=notification_handler - ) - mock_gates_init.assert_called_once_with( ANY, logger=logger, @@ -93,9 +76,7 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre @mark.asyncio @patch('model_manager.activities.activities.Postgres', return_value=MagicMock()) @patch('model_manager.activities.activities.MLFlow', return_value=MagicMock()) -@patch('model_manager.activities.activities.OPC', return_value=MagicMock()) -async def test_shutdown(mock_opc_init, - _mock_mlflow_init, mock_postgres_init): +async def test_shutdown(_mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', 'port': 5432, @@ -113,23 +94,15 @@ async def test_shutdown(mock_opc_init, '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 ) await 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 deleted file mode 100644 index dcff019..0000000 --- a/tests/laborious/activities/test_opc.py +++ /dev/null @@ -1,369 +0,0 @@ -from unittest.mock import patch, MagicMock, ANY, call, AsyncMock -from pandas import DataFrame -from pytest import fixture, mark -import pytest_asyncio -from sientia_do.notifications.models import NotificationLevel - -from model_manager.activities.opc import OPC - -metadata = { - "metadata": { - "model_id": "test_model", - "model_name": "test_model", - "workflow_name": "test_workflow", - "schema_name": "test_schedule", - }, -} - - -def test__init__(): - servers = { - 'server1': 'config' - } - opc = OPC( - opc_servers=servers, - logger=MagicMock(), - notification_handler=MagicMock() - ) - - assert opc.opc_servers == servers - assert opc.opc_repository == {} - - -@mark.asyncio -@patch("model_manager.activities.opc.OpcRepository") -@patch("model_manager.activities.opc.OPC.send_notification") -async def test_init_opc(mock_send_notification, mock_opc_repository): - mock_logger = MagicMock() - server1 = MagicMock( - connect=AsyncMock(return_value=(True, {})), - write_data=AsyncMock(return_value=(True, {})) - ) - server2 = MagicMock( - connect=AsyncMock(return_value=(True, {})), - write_data=AsyncMock(return_value=(True, {})) - ) - server3 = MagicMock( - connect=AsyncMock(return_value=(False, { - 'notification_id': 'OPC_CONNECTION_ERROR_server3', - 'message': 'Failed to connect to OPC server: Test error', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': 'Test error' - })), - write_data=AsyncMock(return_value=(True, {})) - ) - mock_opc_repository.side_effect = [server1, server2, server3] - mock_notification_handler = MagicMock() - servers = { - 'server1': { - 'id': 'server1', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - }, - 'server2': { - 'id': 'server2', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - }, - 'server3': { - 'id': 'server3', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - } - } - opc = OPC( - opc_servers=servers, - logger=mock_logger, - notification_handler=mock_notification_handler - ) - await opc.init_opc() - - assert opc.opc_servers == servers - assert opc.logger == mock_logger - assert opc.notification_handler == mock_notification_handler - assert opc.opc_repository['server1'] == server1 - assert opc.opc_repository['server2'] == server2 - - mock_opc_repository.assert_has_calls([ - call( - id="server1", - url="http://localhost:8080", - logger=mock_logger, - server_uri="opc.tcp://localhost:4840", - cert_path="", - private_key_path="", - server_cert_path="", - notification_handler=mock_notification_handler, - reconnection_interval=60, - pod_id='localhost' - ), - ]) - mock_opc_repository.assert_has_calls([ - call( - id="server2", - url="http://localhost:8080", - logger=mock_logger, - server_uri="opc.tcp://localhost:4840", - cert_path="", - private_key_path="", - server_cert_path="", - notification_handler=mock_notification_handler, - reconnection_interval=60, - pod_id='localhost' - ) - ]) - - server1.connect.assert_called_once() - server2.connect.assert_called_once() - - mock_send_notification.assert_has_calls([ - call( - metadata={ - 'model_id': '-', - 'model_name': '-', - 'workflow_name': '-', - 'schedule_name': 'INITIALIZATION' - }, - notification_id="OPC_CONNECTION_ERROR_server3", - message="Failed to connect to OPC server: Test error", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - ]) - - -@pytest_asyncio.fixture -@patch("model_manager.activities.opc.OpcRepository") -async def opc(mock_opc_repository): - servers = { - 'server1': { - 'id': 'server1', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - } - } - - mock_opc_repository.return_value.write_data = AsyncMock( - return_value=(True, {}) - ) - mock_opc_repository.return_value.connect = AsyncMock( - return_value=(True, {}) - ) - opc = OPC( - opc_servers=servers, - logger=MagicMock(), - notification_handler=MagicMock() - ) - await opc.init_opc() - opc.send_notification = MagicMock() - return opc - - -WRITE_DATA_CASES = [ - ('tag1', 'int', 50), - ('tag2', 'float', 50.5), - ('tag3', 'bool', True), - ('tag4', 'string', 'test'), -] - - -@mark.parametrize('tag,data_type,data', WRITE_DATA_CASES) -@mark.asyncio -async def test_write_data_success(opc, tag, data_type, data): - result = await opc.write_data(server_id='server1', tag=tag, data=data, - data_type=data_type, tag_type='prediction', metadata=metadata) - assert result is True - opc.opc_repository['server1'].write_data.assert_called_once_with( - tag, data, data_type, opc.logger, metadata) - - -@mark.asyncio -async def test_write_data_failed(opc): - opc.opc_repository['server1'].write_data.return_value = (False, { - 'notification_id': 'OPC_WRITE_DATA_ERROR_server1', - 'message': 'Failed to write data to OPC server: Test error', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': 'Test error' - }) - - result = await opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) - assert result is False - - opc.send_notification.assert_called_once_with( - metadata=metadata, - notification_id="OPC_WRITE_DATA_ERROR_server1", - message="Failed to write data to OPC server: Test error", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - - -@mark.asyncio -async def test_write_data_exception(opc): - opc.opc_repository['server1'].write_data.side_effect = Exception( - "Test error") - - try: - await opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) - - except Exception: - opc.send_notification.assert_called_once_with( - metadata=metadata, - notification_id="WRITE_OPC_PREDICTION_ERROR", - message="Error writing data to OPC server: Test error", - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - - else: - assert False, "Expected an exception to be raised" - - -@mark.asyncio -async def test_write_opc_data_success(opc): - # Arrange - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_output_config': { - 'server1': { - 'prediction_tags': { - 'tag1': {'data_type': 'float'} - }, - 'confidence_tags': { - 'tag2': {'data_type': 'float'} - } - } - } - } - - # Act - opc.write_data = AsyncMock(return_value=True) - opc.process_confidence = MagicMock(return_value={'data': 'data'}) - output = await opc.write_opc_data(input_data) - - # Assert - assert output == {'data': 'data'} - opc.write_data.assert_has_calls([ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'] - )]) - opc.write_data.assert_has_calls([ - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'] - ) - ]) - assert opc.write_data.call_count == 2 - - -@mark.asyncio -async def test_write_opc_data_empty_config(opc): - # Arrange - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_servers': ['server1'], - 'opc_output_config': { - 'server1': { - 'prediction_tags': {}, - 'confidence_tags': {} - } - } - } - - # Act - await opc.write_opc_data(input_data) - - # Assert - opc.opc_repository['server1'].write_data.assert_not_called() - - -@mark.asyncio -async def test_write_opc_data_no_validate_server(opc): - opc.validate_server = MagicMock(return_value=False) - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_output_config': { - 'server1': { - 'prediction_tags': { - 'tag1': {'data_type': 'float'} - }, - 'confidence_tags': { - 'tag2': {'data_type': 'float'} - } - } - } - } - - # Act - await opc.write_opc_data(input_data) - - # Assert - opc.opc_repository['server1'].write_data.assert_not_called() - - -@mark.parametrize('data,success,expected', [ - (DataFrame({'prediction_confidence': [0]}), True, 0), - (DataFrame({'prediction_confidence': [0]}), False, 12), -]) -def test_process_confidence(opc, data, success, expected): - # Act - result = opc.process_confidence(data, success, metadata) - - # Assert - assert result['prediction_confidence'][0] == expected - - -def test_validate_server(opc): - assert opc.validate_server('server1', metadata) is True - assert opc.validate_server('server2', metadata) is False - - -@mark.asyncio -async def test_shutdown(opc): - opc.opc_repository['server1'].disconnect = AsyncMock(return_value=True) - await opc.shutdown() - opc.opc_repository['server1'].disconnect.assert_called_once() diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py deleted file mode 100644 index 9795a10..0000000 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ /dev/null @@ -1,388 +0,0 @@ -import pytest -from unittest.mock import AsyncMock, Mock, patch, MagicMock, ANY, call -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from model_manager.utils.repository.opc_repository import OpcRepository -from sientia_do.notifications.models import NotificationLevel -from datetime import datetime - - -@pytest.fixture -def mock_logger(): - return Mock() - - -@pytest.fixture -def opc_repository(mock_logger): - return OpcRepository( - id="test_repo", - url="opc.tcp://localhost:4840", - logger=mock_logger, - notification_handler=Mock(), - reconnection_interval=60, - server_uri="urn:test:server", - cert_path="/path/to/cert.pem", - private_key_path="/path/to/key.pem", - server_cert_path="/path/to/server_cert.pem" - ) - - -@pytest.fixture -def mock_client(): - with patch('model_manager.utils.repository.opc_repository.Client') as mock: - client_instance = AsyncMock() - mock.return_value = client_instance - yield client_instance - - -metadata = { - "metadata": { - "model_id": "test_model", - "model_name": "test_model", - "workflow_name": "test_workflow", - "schema_name": "test_schedule", - }, -} - - -def test_init(opc_repository): - assert opc_repository.id == "test_repo" - assert opc_repository.url == "opc.tcp://localhost:4840" - assert opc_repository.server_uri == "urn:test:server" - assert opc_repository.cert_path == "/path/to/cert.pem" - assert opc_repository.private_key_path == "/path/to/key.pem" - assert opc_repository.server_cert_path == "/path/to/server_cert.pem" - assert opc_repository.reconnection_interval == 60 - assert opc_repository.client is None - assert opc_repository.last_reconnection_time is None - assert opc_repository.error_count == 0 - - -@pytest.mark.asyncio -async def test_set_security(opc_repository, mock_client): - opc_repository.client = mock_client - await opc_repository.set_security() - - mock_client.application_uri = "urn:test:server" - mock_client.set_security.assert_called_once_with( - SecurityPolicyBasic256, - certificate="/path/to/cert.pem", - private_key="/path/to/key.pem", - server_certificate="/path/to/server_cert.pem" - ) - assert mock_client.secure_channel_timeout == 10000000 - assert mock_client.session_timeout == 10000000 - - -@pytest.mark.asyncio -async def test_set_security_missing_certificates(opc_repository): - opc_repository.cert_path = None - opc_repository.private_key_path = None - - try: - await opc_repository.set_security() - except ValueError as e: - assert str( - e) == "Certificate and private key paths must be provided for secure connection." - - -@pytest.mark.asyncio -async def test_connect_with_security(opc_repository, mock_client): - opc_repository.try_connect = AsyncMock(return_value=(True, {})) - result = await opc_repository.connect() - - opc_repository.try_connect.assert_called_once() - assert opc_repository.client == mock_client - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_connect_without_security(opc_repository, mock_client): - opc_repository.cert_path = None - opc_repository.try_connect = AsyncMock(return_value=(True, {})) - opc_repository.set_security = AsyncMock() - result = await opc_repository.connect() - - opc_repository.try_connect.assert_called_once() - opc_repository.set_security.assert_not_called() - assert opc_repository.client == mock_client - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_try_connect_success(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.client = AsyncMock() - result = await opc_repository.try_connect() - - opc_repository.client.connect.assert_called_once() - assert opc_repository.last_reconnection_time is not None - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_try_connect_fail(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.client = MagicMock() - opc_repository.client.connect.side_effect = Exception("Test error") - - is_connected, error_data = await opc_repository.try_connect() - - opc_repository.client.connect.assert_called_once() - assert is_connected is False - assert error_data['notification_id'] == f"OPC_CONNECTION_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to connect to OPC server: Test error" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None - - -@pytest.mark.asyncio -async def test_disconnect(opc_repository, mock_client): - opc_repository.client = mock_client - await opc_repository.disconnect() - - mock_client.disconnect.assert_called_once() - assert opc_repository.client is None - - -@pytest.mark.asyncio -async def test_disconnect_no_client(opc_repository): - opc_repository.client = None - assert await opc_repository.disconnect() is None - - -@pytest.mark.asyncio -async def test_disconnect_error(opc_repository, mock_client): - opc_repository.client = mock_client - mock_client.disconnect.side_effect = Exception("Test error") - await opc_repository.disconnect() - - opc_repository.logger.custom_error.assert_called_once_with( - "Failed to disconnect from OPC server: Test error", - ANY - ) - assert opc_repository.client is None - - -@pytest.mark.asyncio -async def test_validate_connection_none_client(opc_repository): - opc_repository.client = None - opc_repository.connect = AsyncMock(return_value=(True, {})) - response = await opc_repository.validate_connection() - assert response == (True, {}) - opc_repository.connect.assert_called_once() - - -@pytest.mark.asyncio -async def test_validate_connection_error_count_disconnect_error(opc_repository): - opc_repository.error_count = 6 - opc_repository.client = AsyncMock() - opc_repository.disconnect = AsyncMock( - side_effect=Exception("Test error") - ) - opc_repository.connect = AsyncMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - assert response == opc_repository.connect.return_value - opc_repository.disconnect.assert_called_once() - opc_repository.connect.assert_called_once() - opc_repository.logger.custom_error.assert_has_calls( - [ - call("Failed to disconnect from OPC server: Test error", ANY), - ] - ) - - -@pytest.mark.asyncio -async def test_validate_connection_error_validate_connection_error(opc_repository): - opc_repository.client = MagicMock( - uaclient=Exception("Test error") - ) - opc_repository.error_count = 0 - - response = await opc_repository.validate_connection() - - assert response == (False, { - "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}", - "message": "Failed to validate connection to OPC server: 'Exception' object has no attribute 'protocol'", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": ANY - }) - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository): - _mock_datetime.now = MagicMock( - return_value=datetime(2025, 1, 1, 0, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = MagicMock() - opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - opc_repository.connect.assert_not_called() - assert response == (False, { - "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}", - "message": f"OPC server {opc_repository.id} is not connected, waiting for next reconnection window...", - "block": "opc_repository", - "level": NotificationLevel.WARNING - }) - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository): - mock_datetime.now = MagicMock( - return_value=datetime(2025, 1, 1, 1, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = AsyncMock() - opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = AsyncMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - opc_repository.connect.assert_called_once() - assert response == opc_repository.connect.return_value - - -@pytest.mark.asyncio -async def test_validate_connection_success(opc_repository): - opc_repository.client = MagicMock() - opc_repository.error_count = 0 - opc_repository.client.uaclient.protocol = MagicMock() - opc_repository.client.uaclient.protocol.state = "open" - - output = await opc_repository.validate_connection() - assert output == (True, {}) - - -@pytest.mark.asyncio -async def test_write_data_validate_connection_do_nothing(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = AsyncMock( - get_node=MagicMock() - ) - mock_node = AsyncMock() - opc_repository.client.get_node.return_value = mock_node - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_write_data_validate_connection_failed(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(False, {})) - opc_repository.client = AsyncMock() - opc_repository.error_count = 0 - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_not_called() - assert result == (False, {}) - - -@pytest.mark.asyncio -async def test_write_data_get_node_failed(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = AsyncMock() - opc_repository.error_count = 0 - opc_repository.client.get_node = MagicMock( - side_effect=Exception("Test error")) - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to get node from OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None - - -@pytest.mark.asyncio -async def test_write_data_invalid_data_type(opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - mock_client.get_node = MagicMock(return_value=mock_node) - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "invalid_type", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Unsupported data type: invalid_type | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data.get('attachment_content') is None - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.metrics') -async def test_write_data(mock_metrics, opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - mock_client.get_node = MagicMock(return_value=mock_node) - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - mock_node.write_value.assert_called_once() - assert result == (True, {}) - - mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.assert_called_once_with( - pod_id=opc_repository.pod_id, - model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], - opc_server_id=opc_repository.id - ) - mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.return_value.inc.assert_called_once_with() - - mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.assert_called_once_with( - pod_id=opc_repository.pod_id, - model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], - opc_server_id=opc_repository.id - ) - mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.return_value.observe.assert_called_once_with( - ANY) - - -@pytest.mark.asyncio -async def test_write_data_write_value_failed(opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - opc_repository.error_count = 0 - mock_client.get_node = MagicMock(return_value=mock_node) - mock_node.write_value.side_effect = Exception("Test error") - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - mock_node.write_value.assert_called_once() - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_DATA_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to write data to OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index 43a2ec9..6179b33 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -1,6 +1,5 @@ from os import environ from model_manager.utils.connectors_config import (build_mlflow_config, - build_opc_config, build_postgres_config, build_mongodb_config) @@ -40,54 +39,6 @@ def test_build_mlflow_config_with_defaults(): assert config['password'] == 'aignosi' -def test_build_opc_config_with_env_vars(): - # Arrange - environ['OPC_CONFIG'] = '{"opc": {"name": "test-opc", "url": "opc.tcp://test:4840"}}' - - # Act - config = build_opc_config() - - # Assert - assert config['opc']['name'] == 'test-opc' - assert config['opc']['url'] == 'opc.tcp://test:4840' - - -def test_build_opc_config_with_individual_env_vars(): - # Arrange - environ.pop('OPC_CONFIG', None) - environ['OPC_ID'] = '1' - environ['OPC_URL'] = 'opc.tcp://test:4840' - environ['OPC_SERVER_URI'] = 'opc.tcp://test:4840' - environ['OPC_RECONNECTION_INTERVAL'] = '300' - - # Act - config = build_opc_config() - - # Assert - assert config['1']['id'] == '1' - assert config['1']['url'] == 'opc.tcp://test:4840' - assert config['1']['server_uri'] == 'opc.tcp://test:4840' - assert config['1']['reconnection_interval'] == 300 - - -def test_build_opc_config_with_defaults(): - # Arrange - environ.pop('OPC_CONFIG', None) - environ.pop('OPC_ID', None) - environ.pop('OPC_URL', None) - environ.pop('OPC_SERVER_URI', None) - environ.pop('OPC_RECONNECTION_INTERVAL', None) - - # Act - config = build_opc_config() - - # Assert - assert config['1']['id'] == '1' - assert config['1']['url'] == 'opc.tcp://localhost:4840' - assert config['1']['server_uri'] == 'opc.tcp://localhost:4840' - assert config['1']['reconnection_interval'] == 120 - - def test_build_postgres_config_with_env_vars(): # Arrange environ['POSTGRES_HOST'] = 'test-host' diff --git a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py index 356c32b..3606d5d 100644 --- a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py +++ b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py @@ -34,8 +34,6 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): "prediction_confidence": 0, "schema": "test_schema", "table_name": "test_table", - "opc_servers": ["test_server"], - "opc_output_config": {"test": "config"}, "prediction_store_policy": "erl:1" } @@ -56,19 +54,6 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): start_to_close_timeout=ANY )]) - workflow_mock.execute_activity_method.assert_has_calls([ - call( - Activities.write_opc_data, - { - 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value, - **metadata - }, - retry_policy=ANY, - start_to_close_timeout=ANY - ) - ]) - workflow_mock.execute_activity_method.assert_has_calls([ call( Activities.export_data_to_postgres, @@ -86,7 +71,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): start_to_close_timeout=ANY )]) - assert workflow_mock.execute_activity_method.call_count == 3 + assert workflow_mock.execute_activity_method.call_count == 2 assert workflow_mock.execute_local_activity_method.call_count == 1 @@ -103,8 +88,6 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction "prediction_confidence": 0, "schema": "test_schema", "table_name": "test_table", - "opc_servers": ["test_server"], - "opc_output_config": {"test": "config"}, "comment": "test_comment" } @@ -125,19 +108,6 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction ) ]) - workflow_mock.execute_activity_method.assert_has_calls([ - call( - Activities.write_opc_data, - { - 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value, - **metadata - }, - retry_policy=ANY, - start_to_close_timeout=ANY - ) - ]) - workflow_mock.execute_activity_method.assert_has_calls([ call( Activities.export_data_to_postgres, diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 46b1565..6922ef1 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -38,7 +38,7 @@ async def test_run(workflow_mock, prediction_process): 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': 'lts:1' } @@ -125,7 +125,7 @@ async def test_run(workflow_mock, prediction_process): 'model_id': 1, 'model_name': 'test_model_name', 'model_config': input_data['model_config'], - 'opc_output_config': input_data['opc_output_config'], + 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'comment': 'Error', @@ -153,7 +153,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -201,7 +201,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -272,7 +272,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -352,7 +352,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -536,7 +536,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'last_timestamp': last_timestamp, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, 'Prediction Process' ) @@ -558,7 +558,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'schema': schema, 'table_name': table_name, 'comment': 'Prediction Process', - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy } ) @@ -590,7 +590,7 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process): 'last_timestamp': last_timestamp, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, "" ) diff --git a/tests/laborious/workflows/test_predictions_batch.py b/tests/laborious/workflows/test_predictions_batch.py index 0d7d477..e71a462 100644 --- a/tests/laborious/workflows/test_predictions_batch.py +++ b/tests/laborious/workflows/test_predictions_batch.py @@ -32,7 +32,7 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch 'query': 'SELECT * FROM test', 'schema': 'test_schema', 'table_name': 'test_table', - 'opc_output_config': 'test_opc_output_config', + 'datetime_columns': ['timestamp', 'created_at'], 'prediction_store_policy': 'erl:1', 'model_config': { @@ -78,7 +78,7 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch }), 'model_config': input_data.get('model_config', {}), 'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']), - 'opc_output_config': input_data.get('opc_output_config', {}), + 'prediction_store_policy': input_data.get('prediction_store_policy', 'erl:1') } diff --git a/validate.sh b/validate.sh new file mode 100755 index 0000000..6d5ab14 --- /dev/null +++ b/validate.sh @@ -0,0 +1,99 @@ +#!/bin/bash +# Model Manager Code Validation Script +# This script runs all code quality checks before committing or deploying + +set -e # Exit on any error + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +NC='\033[0m' # No Color + +echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" +echo -e "${BLUE}║ Model Manager - Code Validation Suite ║${NC}" +echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" +echo "" + +# Check if virtual environment is activated +if [[ -z "${VIRTUAL_ENV}" ]] && [[ -z "${CONDA_DEFAULT_ENV}" ]]; then + echo -e "${YELLOW}⚠️ Warning: No virtual environment detected${NC}" + echo -e "${YELLOW} Consider activating your venv/conda environment${NC}" + echo "" +fi + +# Function to run a validation step +run_step() { + local step_name=$1 + local step_command=$2 + + echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo -e "${BLUE}▶ ${step_name}${NC}" + echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + + if eval "$step_command"; then + echo -e "${GREEN}✅ ${step_name} - PASSED${NC}" + echo "" + return 0 + else + echo -e "${RED}❌ ${step_name} - FAILED${NC}" + echo "" + return 1 + fi +} + +# Track failures +FAILED_STEPS=() + +# Step 1: Code Formatting Check (Ruff) +if ! run_step "1. Code Formatting (Ruff)" "ruff format --check model_manager/ tests/"; then + FAILED_STEPS+=("Code Formatting") +fi + +# Step 2: Linting (Ruff) +if ! run_step "2. Code Linting (Ruff)" "ruff check model_manager/ tests/"; then + FAILED_STEPS+=("Linting") +fi + +# Step 3: Type Checking (mypy) +if ! run_step "3. Type Checking (mypy)" "mypy model_manager/"; then + FAILED_STEPS+=("Type Checking") +fi + +# Step 4: Security Analysis (Bandit) +if ! run_step "4. Security Analysis (Bandit)" "bandit -r model_manager/ -ll -q"; then + FAILED_STEPS+=("Security Analysis") +fi + +# Step 5: Unit Tests (pytest) +if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=model_manager --cov-report=term-missing --cov-fail-under=70 -q"; then + FAILED_STEPS+=("Unit Tests") +fi + +# Summary +echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" +echo -e "${BLUE}║ Validation Summary ║${NC}" +echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" +echo "" + +if [ ${#FAILED_STEPS[@]} -eq 0 ]; then + echo -e "${GREEN}✅ All validation checks passed!${NC}" + echo -e "${GREEN} Your code is ready for commit/deployment.${NC}" + echo "" + exit 0 +else + echo -e "${RED}❌ Validation failed for the following steps:${NC}" + for step in "${FAILED_STEPS[@]}"; do + echo -e "${RED} • ${step}${NC}" + done + echo "" + echo -e "${YELLOW}💡 Tips:${NC}" + echo -e "${YELLOW} • Run 'ruff format model_manager/ tests/' to auto-fix formatting${NC}" + echo -e "${YELLOW} • Run 'ruff check --fix model_manager/ tests/' to auto-fix linting issues${NC}" + echo -e "${YELLOW} • Review mypy errors and add type hints where needed${NC}" + echo -e "${YELLOW} • Check bandit warnings for security issues${NC}" + echo -e "${YELLOW} • Fix failing tests or improve test coverage${NC}" + echo "" + exit 1 +fi diff --git a/values.yaml b/values.yaml index 744c0c0..fc71161 100644 --- a/values.yaml +++ b/values.yaml @@ -180,11 +180,6 @@ env: - name: MLFLOW_PASSWORD value: "1L0FP50j3ncp123" - - name: OPC_ID - value: "1" - - name: OPC_URL - value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840" - - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka.kafka.svc.cluster.local:9092"