Merge pull request #18 from Aignosi/SIENTIAPDE-1084-ajustar-documentacao

SIENTIAPDE-1084: Refactor and Enhance README.md and Project Configuration

.gitignore

@@ -39,3 +39,5 @@ htmlcov/
 .coverage
 
 git_log
+
+.env

Dockerfile

@@ -1,31 +0,0 @@
-# Use uma imagem base Python
-FROM python:3.11-slim
-
-# Instale git e outras dependências do sistema
-RUN apt-get update && apt-get install -y git \
-    && apt-get install -y build-essential python3-dev \
-    && apt-get install -y vim \
-    && rm -rf /var/lib/apt/lists/*
-
-# Defina o diretório de trabalho
-WORKDIR /app
-
-# Copie os arquivos do projeto
-COPY . /app
-
-RUN pip install --upgrade pip setuptools wheel
-
-# Install the required packages
-# Add github to known hosts
-# This is needed for SSH to work
-# The SSH key will NOT remain in the image
-# IMPORTANT: this block requires BuildKit
-# and the --ssh flag during docker build
-RUN --mount=type=ssh \
-    mkdir -p ~/.ssh && \
-    ssh-keyscan github.com >> ~/.ssh/known_hosts && \
-    pip install --no-cache-dir -r requirements.txt
-
-
-# Defina o comando para executar o worker
-CMD ["python", "-m", "scouter.worker.worker"]

README.md

@@ -1,101 +1,567 @@
 # Sientia DataOps Scouter
 
-The Sientia DataOps Scouter is a Temporal-based workflow application that processes industrial data from OPC collectors. It aggregates and filters data received from OPC collectors via Kafka topics, direct access to the OPC server, or active trigger (real time applications). The processed data is then stored or forwarded for further analysis.
+A high-performance, scalable data processing system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities.
 
-## Key Features
+## Features
 
-- Data ingestion from multiple sources:
+### Core Functionality
-  - OPC collectors through Kafka
+- **Multi-Source Data Ingestion**: Support for Kafka topics, direct OPC server access, and real-time triggers
-  - Direct access to OPC servers
+- **Temporal Workflow Orchestration**: Robust workflow management with automatic retry policies and fault tolerance
-  - Real-time triggers for immediate processing
+- **Data Quality Gates**: Configurable filtering for null values, out-of-bounds data, and custom validation rules
-- Data aggregation and filtering
+- **Time-Series Aggregation**: Flexible aggregation functions (average, median, max, min, latest) with configurable parameters
-- Workflow orchestration using Temporal.io
+- **Multi-Database Integration**: PostgreSQL for persistent storage, Redis for caching, MongoDB for data retrieval
-- Integration with Redis for caching and PostgreSQL for storage
+- **Real-time Monitoring**: Prometheus metrics and comprehensive logging for operational visibility
-- Scalable deployment using Kubernetes
 
-## Workflows
+### Advanced Capabilities
+- **Incremental Data Processing**: Timestamp-based data loading to avoid reprocessing
+- **Configurable Data Retention**: Redis-based temporary storage with TTL management
+- **Notification System**: Integrated alerting and notification management via MongoDB
+- **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support
+- **Test Data Generation**: Built-in fake data generation for development and testing
 
-### Core Scouter
+# Architecture
 
-The Core Scouter is the main workflow processes the received data. Steps:
+The Scouter system uses a Temporal-based workflow architecture with clear separation of concerns:
 
-- data_quality_gate: Filters the data received from the OPC collector.
+### Key Components
-- aggregate_data: Aggregates the data received from the OPC collector.
-- group_and_hold_data: Groups the data received from the OPC collector.
-- export_data_to_postgres: Exports the data received from the OPC collector to PostgreSQL.
 
-### Scouter
+- **Worker**: Main application orchestrator managing Temporal workers and task queues
+- **Workflows**: Temporal workflow definitions for data processing orchestration
+- **Activities**: Temporal activities implementing data processing operations
+- **Data Services**: Database connectors and data access layer
+- **Quality Filters**: Configurable data validation and filtering mechanisms
 
-The Scouter is the batch basic workflow that extracts data from the source and processes it using the Core Scouter workflow. Steps:
+## 🔄 Workflows
 
-- load_from_kafka: Loads data from a kafka topic.
+### 1. Scouter Workflow (`scouter.py`)
-- core_scouter: Processes the data using the Core Scouter workflow.
 
-#### Workflow inputs:
+The **Scouter** workflow is the main entry point for data processing pipelines. It orchestrates the complete data ingestion process and implements a robust incremental data processing pattern.
 
-- `topic` (str): Kafka topic name where data is received
+#### Purpose
-- `schedule_name` (str): Name of the schedule that triggers the workflow
+- **Data Ingestion Orchestration**: Coordinates data loading from MongoDB collections
-- `model_name` (str): Name of the model being used for processing
+- **Timestamp Management**: Tracks last processed timestamps to enable incremental processing
-- `model_id` (int): Unique identifier for the model
+- **Workflow Delegation**: Delegates actual data processing to the CoreScouter workflow
-- `trigger_laborious` (bool): Flag indicating if laborious direct processing is required (real time applications)
+- **Data Continuity**: Ensures no data is lost or reprocessed between executions
-- `filters` (dict): Dictionary containing data filtering rules
-  - `NULL_VALUES_FILTER`: Configuration for handling null values
-    - `policy`: Policy for null values ("KEEP" or "DISCARD")
-  - `OUT_OF_BOUNDS_FILTER`: Configuration for handling out-of-bounds values
-    - `policy`: Policy for out-of-bounds values ("KEEP" or "DISCARD")
-- `schema` (str): Database schema name where data will be stored
-- `table_name` (str): Name of the table where data will be stored
-- `retention_time` (int): Time in seconds that data will be retained in Redis
-- `model_tags` (dict): Configuration for different OPC tags
-  - The key is the tag name and the value is a dictionary containing:
-    - `data_range`: List of two numbers [min, max] defining valid data range
-    - `aggr_function`: Aggregation function to use ("lts", "mdn", "avg", "max", "min")
 
-## Fake Data
+#### Execution Flow
+1. **Timestamp Retrieval**: Gets the last processed timestamp from Redis for the specific workflow and schedule
+2. **Data Loading**: Loads new data from MongoDB since the last timestamp using the collection name `raw_{schedule_name}`
+3. **Timestamp Update**: Updates the last processed timestamp with the most recent data point
+4. **Data Processing**: Delegates data processing to the CoreScouter child workflow
+5. **Metadata Management**: Maintains workflow execution metadata throughout the process
 
-The Fake Data activity is used to generate fake data for testing purposes. Steps:
+#### Key Features
+- **Incremental Processing**: Only processes new data since last execution
+- **Automatic Retry**: Implements Temporal retry policies for fault tolerance
+- **Timeout Management**: 60-second timeout for all activities
+- **Error Handling**: Comprehensive error handling with notification integration
 
-- generate_and_send_data: Generates fake data and sends it to a kafka topic.
+#### Input Parameters
+```json
+{
+  "topic": "sensor_data_topic",
+  "schedule_name": "hourly_collection",
+  "model_name": "temperature_sensors",
+  "model_id": "temp_001",
+  "trigger_laborious": false,
+  "filters": {...},
+  "schema": "sensor_data",
+  "table_name": "temperature_readings",
+  "retention_time": 3600,
+  "model_tags": {...}
+}
+```
 
-#### Workflow inputs:
+#### Architecture
 
-- `topic` (str): Kafka topic name where data is received
+```mermaid
+flowchart LR
+    A[1. get_last_data_timestamp] --> B[2. load_latest_data] --> C[3. put_last_data_timestamp] --> D[4. core_scouter 🔃]
     
-## Environment variables
+    A -.-> Redis[(Redis)]
+    B -.-> MongoDB[(MongoDB)]
+    C -.-> Redis
+```
 
-- `POSTGRES_HOST`
-- `POSTGRES_PORT`
-- `POSTGRES_USER`
-- `POSTGRES_PASSWORD`
-- `POSTGRES_DBNAME`
-- `POSTGRES_MIN_CONNECTIONS`
-- `POSTGRES_MAX_CONNECTIONS`
 
-- `KAFKA_BOOTSTRAP_SERVERS`
+### 2. CoreScouter Workflow (`core_scouter.py`)
-- `KAFKA_POLLING_TIME`
 
-- `REDIS_HOST`
+The **CoreScouter** workflow implements the core data processing pipeline for industrial time-series data. It handles data quality validation, aggregation, and export operations.
-- `REDIS_PORT`
-- `REDIS_USERNAME`
-- `REDIS_PASSWORD`
 
-- `LOG_LEVEL`
+#### Purpose
-- `PROJECT_NAME`
+- **Data Quality Validation**: Applies configurable filters for data integrity
+- **Time-Series Aggregation**: Groups and aggregates data using specified functions
+- **Data Organization**: Groups data by tags and applies retention policies
+- **Persistent Storage**: Exports processed data to PostgreSQL
+- **Metrics Collection**: Records processing metrics for monitoring
 
-- `TEMPORAL_HOST`
+#### Execution Flow
-- `TEMPORAL_NAMESPACE`
+1. **Data Quality Gate**: Applies configured filters (null values, out-of-bounds, custom rules)
+2. **Data Aggregation**: Groups data by tag and name, applies aggregation functions
+3. **Data Grouping**: Organizes data and stores temporarily in Redis with TTL
+4. **Data Export**: Persists processed data to PostgreSQL database
+5. **Metrics Recording**: Writes processing metrics for operational visibility
 
-## Application deployment
+#### Aggregation Functions
+- **`lts`**: Latest value (most recent data point)
+- **`avg`**: Average of all values in the group
+- **`mdn`**: Median of all values in the group
+- **`max`**: Maximum value in the group
+- **`min`**: Minimum value in the group
 
-The application can be deployed using the following command:
+#### Key Features
+- **Configurable Quality Gates**: Multiple filter types with policy-based configuration
+- **Flexible Aggregation**: Tag-specific aggregation function configuration
+- **Batch Processing**: Efficient handling of large datasets
+- **Asynchronous Export**: Non-blocking data export operations
+- **Comprehensive Monitoring**: Detailed metrics and error reporting
+
+#### Input Parameters
+```json
+{
+  "metadata": {...},
+  "workflow_name": "scouter",
+  "schedule_name": "hourly_collection",
+  "model_name": "temperature_sensors",
+  "model_id": "temp_001",
+  "data": {...},
+  "trigger_laborious": false,
+  "filters": {...},
+  "schema": "sensor_data",
+  "table_name": "temperature_readings",
+  "retention_time": 3600,
+  "model_tags": {
+    "Temperature": {
+      "data_range": [-50, 150],
+      "aggr_function": "avg",
+      "frequency": "60000"
+    }
+  }
+}
+```
+
+#### Architecture
+
+```mermaid
+flowchart LR
+    A[1. data_quality_gate] --> B[2. aggregate_data] --> C[3. group_and_hold_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics]
+    
+    B -.-> Redis1[(Redis)]
+    C -.-> Redis2[(Redis)] 
+    D -.-> PostgreSQL[(PostgreSQL)]
+    E -.-> Metrics[Prometheus]
+```
+
+### 3. FakeData Workflow (`fake_data.py`)
+
+The **FakeData** workflow generates synthetic industrial sensor data for testing and development purposes. It's designed to simulate realistic data flows without requiring actual industrial data sources.
+
+#### Purpose
+- **Test Data Generation**: Creates realistic sensor data for development and testing
+- **Pipeline Validation**: Tests data processing workflows with known data
+- **Load Testing**: Generates configurable data volumes for performance testing
+- **Demonstration**: Shows data flow patterns and processing capabilities
+
+#### Execution Flow
+1. **Data Generation**: Creates synthetic sensor readings with realistic values
+2. **Kafka Publishing**: Sends generated data to specified Kafka topics
+3. **Quality Assurance**: Ensures data format consistency and completeness
+4. **Monitoring**: Tracks generation and publishing metrics
+
+#### Key Features
+- **Realistic Data**: Generates data within realistic industrial ranges
+- **Configurable Volume**: Adjustable message counts for different testing scenarios
+- **Random Variation**: Includes realistic data variations and occasional null values
+- **Kafka Integration**: Direct integration with Kafka for data streaming
+- **Error Handling**: Comprehensive error handling and logging
+
+#### Input Parameters
+```json
+{
+  "topic": "test_sensor_data",
+  "metadata": {...},
+  "num_messages": 100
+}
+```
+
+#### Architecture
+
+```mermaid
+flowchart TB
+    subgraph workflow [" "]
+        A[1. generate_and_send_data]
+    end
+    
+    subgraph services [" "]
+        Kafka[(Kafka)]
+    end
+    
+    A -.-> Kafka
+    
+    style workflow fill:none,stroke:none
+    style services fill:none,stroke:none
+```
+
+## 📋 Prerequisites
+
+- Python 3.11+
+- Temporal server/cluster
+- PostgreSQL database
+- Redis server
+- MongoDB server
+- Kafka cluster (for data ingestion)
+
+**Note**: External dependencies must be available either through:
+- Kubernetes cluster deployment
+- Docker Compose setup
+- Cloud-managed services
+- Local installations
+
+## 🚀 Installation
+
+### Local Development Setup
+
+1. **Clone the repository**
+   ```bash
+   git clone <repository-url>
+   cd sientia-dataops-scouter
+   ```
+
+2. **Create virtual environment**
+   ```bash
+   python3.11 -m venv venv
+   source ./venv/bin/activate
+   ```
+
+3. **Install dependencies**
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+4. **Create environment configuration file**
+   ```bash
+   cp .env.example .env
+   # Edit .env with your connection details
+   ```
+
+5. **Configure external dependencies**
+   
+   You'll need to set up port forwarding or connections to external services. For example:
    
    ```bash
-helm upgrade --install sientia-dataops-opc-ingestor sientia/sientia-module -n sientia-opc --create-namespace -f ./values.yaml
+   # Port forwarding from Kubernetes cluster
+   kubectl port-forward svc/redis-master 6379:6379
+   kubectl port-forward svc/mongodb 27017:27017
+   kubectl port-forward svc/kafka 9092:9092
+   
+   # Or connect to external services
+   # Ensure services are accessible on localhost with appropriate ports
    ```
 
-#PR shortcut
+## 📦 How to Run
+
+### Running the Scouter Application
+
+Use the provided script to run the application locally:
+
+```bash
+# Make script executable (first time only)
+chmod +x run_local.sh
+
+# Run the application
+./run_local.sh
 ```
-git log origin/main..HEAD --no-merges > git_log
+
+The script will:
+- Activate the virtual environment
+- Load environment variables from `.env`
+- Start the scouter worker application
+
+### Running Tests and Coverage
+
+Use the provided script to run tests with coverage:
+
+```bash
+# Make script executable (first time only)
+chmod +x run_coverage.sh
+
+# Run tests with coverage
+./run_coverage.sh
 ```
-Prompt:
+
-Write a summary of PR changes in markdown. Be objective and direct. Write to file
+The script will:
+- Activate the virtual environment
+- Run pytest with coverage reporting
+- Generate HTML coverage report
+- Open the coverage report in your browser
+
+### Manual Test Execution
+
+You can also run tests manually:
+
+```bash
+# Activate virtual environment
+source ./venv/bin/activate
+
+# Run all tests
+pytest
+
+# Run with coverage
+pytest --cov=scouter --cov-report=html
+
+# Run specific test categories
+pytest tests/activities/
+pytest tests/workflow/
+```
+
+### Manual Application Execution
+
+For manual execution without scripts:
+
+```bash
+# Activate virtual environment
+source ./venv/bin/activate
+
+# Load environment variables (if using .env file)
+if [ -f .env ]; then
+    export $(cat .env | grep -v '^#' | xargs)
+fi
+
+# Start the scouter worker
+python -m scouter.worker.worker
+```
+
+### Environment Configuration
+
+Before running the application, ensure your `.env` file contains the necessary configuration.
+
+## 🧪 Testing
+
+### Test Structure
+```
+tests/
+├── activities/           # Activity implementation tests
+├── workflow/            # Workflow orchestration tests
+├── utils/               # Utility function tests
+└── integration/         # End-to-end workflow tests
+```
+
+### Test Execution
+```bash
+# Install test dependencies
+pip install pytest pytest-cov pytest-asyncio
+
+# Run tests with coverage
+pytest --cov=scouter --cov-report=html
+
+# Run specific test modules
+pytest tests/activities/test_redis.py
+pytest tests/workflow/test_scouter.py
+```
+
+## 📊 Monitoring and Metrics
+
+The Scouter system exposes comprehensive Prometheus metrics:
+
+### Application Metrics
+- `app_up`: Application health status (1=healthy, 0=unhealthy)
+- `scouter_laborious_data_written_count`: Data export operation count
+- `scouter_tag_changes_monitor`: Tag value change monitoring
+
+### Temporal Metrics
+- Workflow execution counts and durations
+- Activity execution success/failure rates
+- Task queue processing metrics
+- Worker health and performance indicators
+
+### Database Metrics
+- Connection pool utilization
+- Query execution times
+- Error rates and retry counts
+
+## ⚙️ Configuration
+
+### Environment Variables
+
+| Variable | Description | Default | Required |
+|----------|-------------|---------|----------|
+| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes |
+| `TEMPORAL_NAMESPACE` | Temporal namespace | `scouter` | No |
+| `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
+| `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes |
+| `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
+| `POSTGRES_PASSWORD` | PostgreSQL password | `sientia` | Yes |
+| `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes |
+| `REDIS_HOST` | Redis hostname | `localhost` | Yes |
+| `REDIS_PORT` | Redis port | `6379` | Yes |
+| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes |
+| `KAFKA_BOOTSTRAP_SERVERS` | Kafka broker addresses | `localhost:9092` | No |
+| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
+| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
+
+### Workflow Configuration
+
+MongoDB pipeline configuration:
+
+#### Scouter Workflow Input Parameters
+
+```json
+{
+  "schedule_name": "scouter-opcua-orchestrated-pipeline",
+  "model_id": "1",
+  "workflow_type": "scouter",
+  "frequency": "5s", # Workflow execution frequency
+  "max_retry_policy": 1, # Maximum number of retries for the workflow
+  "read_tags": [
+    {
+      "tag_name": "Counter",
+      "server_id": "1",
+      "aggr_func": "avg",
+      "tag_address": "ns=2;i=2",
+      "frequency": "15000", # Tag expected frequency (used in Ingestor)
+      "data_range": [
+        -100,
+        100
+      ]
+    },
+    {
+      "tag_name": "Rollout",
+      "server_id": "1",
+      "aggr_func": "mdn",
+      "tag_address": "ns=2;i=3",
+      "frequency": "15000",
+      "data_range": [
+        -100,
+        100
+      ]
+    }
+  ],
+  "filters": [
+    {
+      "filter_name": "OUT_OF_BOUNDS_FILTER",
+      "policy": "DISCARD"
+    },
+    {
+      "filter_name": "NULL_VALUES_FILTER",
+      "policy": "DISCARD"
+    }
+  ],
+  "tag_retention_minutes": 60, # Time that tag data is cached in Redis
+  "active": true,
+  "updated_at": {
+    "$date": "2025-08-13T18:35:01.600Z"
+  }
+}
+```
+
+## 🔧 Development
+
+### Project Structure
+```
+scouter/
+├── activities/              # Temporal activity implementations
+│   ├── activities.py       # Main activities orchestrator
+│   ├── redis.py           # Redis operations
+│   ├── gates.py            # Data quality gates
+│   ├── mongodb.py          # MongoDB operations
+│   └── faker.py            # Test data generation
+├── workflow/                # Temporal workflow definitions
+│   ├── scouter.py          # Main data ingestion workflow
+│   ├── fake_data.py        # Test data workflow
+│   └── sub_workflows/      # Sub-workflow implementations
+│       └── core_scouter.py # Core data processing workflow
+├── worker/                  # Worker implementation
+│   └── worker.py           # Main worker orchestrator
+├── utils/                   # Utility functions
+│   ├── connectors_config.py # Database configuration
+│   └── quality/            # Data quality filters
+├── metrics.py               # Prometheus metrics definitions
+└── __init__.py
+```
+
+### Adding New Features
+
+1. **Follow Temporal patterns** for new workflows and activities
+2. **Add comprehensive docstrings** for all public methods
+3. **Include Prometheus metrics** for monitoring
+4. **Add unit tests** for new functionality
+5. **Update this README** with new features and configuration
+
+## 🐛 Troubleshooting
+
+### Common Issues
+
+1. **Temporal Connection Failures**
+   - Verify Temporal server is running and accessible
+   - Check namespace configuration and permissions
+   - Review server logs for connection issues
+
+2. **Database Connection Issues**
+   - Verify all database services are running
+   - Check connection credentials and network access
+   - Ensure proper connection pool configuration
+
+3. **Workflow Execution Failures**
+   - Review activity error logs and notifications
+   - Check data quality filter configurations
+   - Verify input data format and required fields
+
+4. **Performance Issues**
+   - Monitor Prometheus metrics for bottlenecks
+   - Review database query performance
+   - Check Temporal worker configuration
+
+### Debug Mode
+
+Enable debug logging by setting the log level:
+```bash
+export LOG_LEVEL=DEBUG
+```
+
+## ⚡ Performance Tuning
+
+### Key Parameters
+
+- **Worker Concurrency**: Adjust `max_concurrent_workflow_tasks` and `max_concurrent_activities`
+- **Connection Pools**: Optimize database connection pool sizes
+- **Data Retention**: Configure Redis TTL based on processing requirements
+- **Batch Sizes**: Adjust data processing batch sizes for optimal throughput
+
+### Scaling Considerations
+
+- **Horizontal Scaling**: Deploy multiple worker instances
+- **Task Queue Distribution**: Use multiple task queues for different workflow types
+- **Database Performance**: Optimize indexes and connection pooling
+- **Kafka Partitioning**: Configure appropriate partition counts for data ingestion
+
+## 🤝 Contributing
+
+1. Fork the repository
+2. Create a feature branch
+3. Make your changes with comprehensive testing
+4. Update documentation and docstrings
+5. Submit a pull request
+
+### Code Quality Standards
+
+- Follow PEP 8 style guidelines
+- Include comprehensive docstrings for all public methods
+- Maintain test coverage above 80%
+- Use type hints where appropriate
+- Follow Temporal.io best practices
+
+## 📄 License
+
+This project is licensed under the terms specified in the LICENSE file.
+
+## 🆘 Support
+
+For support and questions:
+- Check the troubleshooting section above
+- Review the metrics and logs for error patterns
+- Open an issue in the project repository
+- Contact the development team
+
+---
+
+**Note**: The Scouter system is designed for production use in industrial data processing environments. Ensure proper security configuration and network isolation for production deployments.

client-schedule.py

@@ -1,42 +0,0 @@
-from temporalio.client import Client, Schedule, ScheduleActionStartWorkflow, ScheduleSpec, ScheduleIntervalSpec
-import asyncio
-from datetime import timedelta
-
-seconds = 5
-
-
-async def main():
-    # Conecta ao servidor Temporal
-    client = await Client.connect("http://localhost:7233")
-
-    for i in range(1, 2):
-        # Define o agendamento para rodar a cada 5 segundos
-        schedule = Schedule(
-            action=ScheduleActionStartWorkflow(
-                'fake_data',  # Nome da classe do workflow no worker.py
-                # Argumento de entrada (ajuste conforme seu workflow)
-                {
-                    'topic': 'fake_data'
-                },
-                # ID que você define aqui
-                id=f"test-{i}-{seconds}",
-                task_queue="fake_data-queue",  # Deve coincidir com o worker
-            ),
-            spec=ScheduleSpec(
-                intervals=[ScheduleIntervalSpec(
-                    every=timedelta(seconds=seconds))]
-            ),
-        )
-
-        # Cria ou atualiza o schedule no Temporal
-        # ID único para o schedule
-        schedule_id = f"test-schedule-c-{i}-every-o-{seconds}s"
-        try:
-            await client.create_schedule(schedule_id, schedule)
-            print(
-                f"Schedule '{schedule_id}' criado com sucesso. Workflow rodará a cada {seconds} segundos.")
-        except Exception as e:
-            print(f"Erro ao criar o schedule: {e}")
-
-if __name__ == "__main__":
-    asyncio.run(main())

coverage.sh

@@ -1 +0,0 @@
-pytest --cov=sientia --cov-report=html && xdg-open htmlcov/index.html

docker-compose.yml

@@ -1,98 +0,0 @@
-version: '3.8'
-
-services:
-  postgres:
-    image: postgres:15
-    container_name: postgres
-    environment:
-      POSTGRES_USER: sientia
-      POSTGRES_PASSWORD: sientia
-      POSTGRES_DB: sientia
-    ports:
-      - "5432:5432"
-    volumes:
-      - ./postgres_data:/var/lib/postgresql/data
-    networks:
-      - sientia-network
-
-  zookeeper:
-    image: confluentinc/cp-zookeeper:7.5.1
-    container_name: zookeeper
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-      ZOOKEEPER_TICK_TIME: 2000
-    ports:
-      - "2181:2181"
-    networks:
-      - sientia-network
-
-  kafka:
-    image: confluentinc/cp-kafka:7.5.1
-    container_name: kafka
-    depends_on:
-      - zookeeper
-    ports:
-      - "9092:9092"
-      - "29092:29092"
-    environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      # Message retention settings (5 minutes)
-      KAFKA_LOG_RETENTION_MINUTES: 5     # 5 minutes
-      KAFKA_LOG_RETENTION_MS: 300000     # 5 minutes in milliseconds
-      KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 30000  # Check every 30 seconds
-    networks:
-      - sientia-network
-
-  kafka-ui:
-    image: provectuslabs/kafka-ui:latest
-    container_name: kafka-ui
-    ports:
-      - "8080:8080"
-    environment:
-      KAFKA_CLUSTERS_0_NAME: local
-      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
-    networks:
-      - sientia-network
-
-  redis:
-    image: redis:latest
-    ports:
-      - "${REDIS_PORT}:6379"
-    networks:
-      - sientia-network
-
-  redis-ui:
-    image: redislabs/redisinsight:latest
-    container_name: redis-ui
-    ports:
-      - "8001:8001"
-    networks:
-      - sientia-network
-    depends_on:
-      - redis
-
-  # scouter:
-  #   build: .
-  #   container_name: scouter
-  #   networks:
-  #     - sientia-network
-  #   env_file:
-  #     - .env
-  #   depends_on:
-  #     - postgres
-  #     - kafka
-  #     - redis
-
-
-networks:
-  sientia-network:
-    driver: bridge
-
-volumes:
-  postgres_data:
-    driver: local

input_sample.json

@@ -1,32 +0,0 @@
-{
-    "topic": "opcua",
-    "schedule_name": "scouter-opcua-pipeline",
-    "model_name": "Demo Model",
-    "model_id": 1,
-    "trigger_laborious": false,
-    "filters": {
-        "NULL_VALUES_FILTER": {
-            "policy": "KEEP"
-        },
-        "OUT_OF_BOUNDS_FILTER": {
-            "policy": "DISCARD"
-        }
-    },
-    "schema": "sientia_data",
-    "table_name": "laborious_data",
-    "retention_time": 3600,
-    "model_tags": {
-        "Counter": {
-            "data_range": [0, 50],
-            "aggr_function": "lts"
-        },
-        "Rollout": {
-            "data_range": [0, 50],
-            "aggr_function": "mdn"
-        },
-        "Square": {
-            "data_range": [0, 100],
-            "aggr_function": "avg"
-        }
-    }
-}

run_coverage.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+# Exit on any error
+set -e
+
+echo "Activating virtual environment..."
+source ./venv/bin/activate
+
+pytest --cov=scouter --cov-report=html
+
+xdg-open htmlcov/index.html

run_local.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+# Exit on any error
+set -e
+
+echo "Activating virtual environment..."
+source ./venv/bin/activate
+
+echo "Loading environment variables from .env..."
+if [ -f .env ]; then
+    export $(cat .env | grep -v '^#' | xargs)
+    echo "Environment variables loaded from .env"
+else
+    echo "Warning: .env file not found. Continuing without environment variables."
+fi
+
+echo "Starting scouter application..."
+python -m scouter.worker.worker

scouter/activities/activities.py

@@ -11,8 +11,21 @@ with workflow.unsafe.imports_passed_through():
     from os import getenv
 
 
-class Activities(Postgres, Redis, Gates, MongoDB,):
+class Activities(Postgres, Redis, Gates, MongoDB):
-    """Activities class that combines multiple services with proper initialization."""
+    """
+    Unified activities class that combines multiple data processing services.
+
+    This class provides a comprehensive interface for all data processing activities
+    by inheriting from specialized service classes. It handles:
+    - PostgreSQL operations for data persistence
+    - Redis operations for caching and temporary storage
+    - Data quality gates and filtering
+    - MongoDB operations for data retrieval
+    - Notification handling and logging
+
+    The class implements the multiple inheritance pattern to provide a unified
+    interface while maintaining separation of concerns across different data services.
+    """
 
     def __init__(self,
                  postgres_config: dict[str, Any],
@@ -20,7 +33,19 @@ class Activities(Postgres, Redis, Gates, MongoDB,):
                  mongodb_config: dict[str, Any],
                  logger: Logger,
                  notification_handler: NotificationHandler):
+        """
+        Initialize the Activities class with all required services.
 
+        Args:
+            postgres_config (dict[str, Any]): PostgreSQL connection configuration.
+                Required fields: host, port, user, password, dbname, min_connections, max_connections
+            redis_config (dict[str, Any]): Redis connection configuration.
+                Required fields: host, port, username, password
+            mongodb_config (dict[str, Any]): MongoDB connection configuration.
+                Required fields: connection_string, database_name
+            logger (Logger): Logger instance for application logging
+            notification_handler (NotificationHandler): Handler for system notifications
+        """
         # Initialize Postgres
         Postgres.__init__(
             self,
@@ -65,5 +90,11 @@ class Activities(Postgres, Redis, Gates, MongoDB,):
         self.pod_id = getenv("HOSTNAME", "localhost")
 
     def shutdown(self):
+        """
+        Gracefully shutdown all service connections.
+
+        This method ensures proper cleanup of database connections and resources
+        to prevent connection leaks and ensure graceful application termination.
+        """
         Postgres.close(self)
         MongoDB.shutdown(self)

scouter/activities/faker.py

@@ -11,8 +11,30 @@ from sientia_do.observability.logger import Logger
 
 
 class Faker(BaseActivity):
+    """
+    Synthetic data generation for testing and development.
+
+    This class generates realistic industrial sensor data for testing purposes.
+    It provides:
+    - Configurable sensor tag simulation
+    - Realistic data value generation
+    - Kafka integration for data publishing
+    - Comprehensive error handling and logging
+
+    The class is designed for development, testing, and demonstration of
+    data processing pipelines without requiring real industrial data sources.
+    """
+
     def __init__(self, bootstrap_servers: str, logger: Logger,
                  notification_handler: NotificationHandler):
+        """
+        Initialize the Faker class with Kafka producer and sensor configuration.
+
+        Args:
+            bootstrap_servers (str): Kafka bootstrap servers configuration
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+        """
         self.producer = KafkaProducer(
             bootstrap_servers=bootstrap_servers,
             value_serializer=lambda v: json.dumps(v).encode('utf-8')
@@ -31,16 +53,28 @@ class Faker(BaseActivity):
         BaseActivity.__init__(self, logger, notification_handler)
 
     @activity.defn(name="generate_and_send_data")
-    async def generate_and_send_data(self, input_data: dict[str, Any]):
+    async def generate_and_send_data(self, input_data: dict[str, Any]) -> None:
         """
-        Generates random data and sends it to a Kafka topic.
+        Generate synthetic sensor data and publish to Kafka topic.
+
+        This activity creates realistic industrial sensor readings and publishes
+        them to the specified Kafka topic. The data includes sensor tags, names,
+        timestamps, and values with configurable message counts.
 
         Args:
-            input_data (dict[str, Any]): The input data containing:
+            input_data (dict[str, Any]): Activity input parameters.
-                topic (str): The Kafka topic to send data to
+                Required fields:
-                num_messages (int, optional): Number of messages to generate. 
+                - topic (str): Kafka topic name for data publication
-                Defaults to random.randint(1, len(self.tags)).
+                - metadata (dict[str, Any], optional): Workflow execution metadata
+                - num_messages (int, optional): Number of messages to generate.
+                  Defaults to random count between 1 and available sensor tags
 
+        Returns:
+            None: This activity publishes data but doesn't return results
+
+        Raises:
+            ValueError: If topic is not specified
+            Exception: If data generation or Kafka publishing fails
         """
         metadata = input_data['metadata']
         topic = input_data.get('topic')

scouter/activities/gates.py

@@ -19,22 +19,52 @@ quality_gate_filters = {
 
 
 class Gates(BaseActivity):
+    """
+    Data quality gates and filtering operations.
+
+    This class implements data quality validation and filtering for industrial
+    time-series data. It provides:
+    - Configurable data quality filters
+    - Data aggregation functions for time-series data
+    - Comprehensive error handling and notification
+    - Metrics collection for quality monitoring
+
+    The class supports multiple aggregation strategies and quality filters to
+    ensure data integrity and enable flexible data processing workflows.
+    """
 
     def __init__(self, logger: Logger, notification_handler: NotificationHandler):
+        """
+        Initialize the Gates class with logging and notification services.
+
+        Args:
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+        """
         BaseActivity.__init__(
             self, logger, notification_handler, set_error_counter=True)
 
     def apply_aggregation(self, group: DataFrame, aggr_function: str,
                           metadata: dict[str, Any]) -> float | None | str:
         """
-        Apply aggregation function to a group of data.
+        Apply aggregation function to a group of time-series data.
+
+        This method applies the specified aggregation function to a group of
+        data points. It handles edge cases and provides comprehensive error
+        reporting for invalid aggregation functions.
 
         Args:
-            group (DataFrame): The group of data to apply the aggregation function to.
+            group (DataFrame): Group of data points to aggregate
-            aggr_function (str): The aggregation function to apply.
+            aggr_function (str): Aggregation function to apply.
+                Supported functions: 'lts' (latest), 'avg' (average), 'mdn' (median),
+                'max' (maximum), 'min' (minimum)
+            metadata (dict[str, Any]): Workflow metadata for error reporting
 
         Returns:
-            float | None | str: The result of the aggregation function.
+            float | None | str: Aggregated value, None if no valid data, or 'continue' for errors
+
+        Raises:
+            NotificationError: If invalid aggregation function is specified
         """
         if len(group) == 1:
             return group['value'].item()
@@ -70,17 +100,23 @@ class Gates(BaseActivity):
     @activity.defn(name="aggregate_data")
     async def aggregate_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
-        Aggregates time series data by tag and name, applying specified
+        Aggregate time-series data by tag and name using specified functions.
-        aggregation functions and taking the latest timestamp.
+
+        This activity processes time-series data by grouping it by tag and name,
+        then applying the configured aggregation functions. It handles data
+        validation and provides comprehensive error reporting.
 
         Args:
-            input_data (dict[str, Any]): The data to aggregate. Contains:
+            input_data (dict[str, Any]): Activity input parameters.
-                data (dict[str, Any]): The time series data.
+                Required fields:
-                model_tags (dict[str, Any]): The tags configuration
+                - data (dict[str, Any]): Time-series data to aggregate
-                containing aggregation functions.
+                - model_tags (dict[str, Any]): Tag configuration with aggregation functions
 
         Returns:
-            dict[str, Any]: The aggregated data.
+            dict[str, Any]: Aggregated data organized by tag and name
+
+        Raises:
+            Exception: If aggregation operation fails
         """
 
         metadata = input_data['metadata']
@@ -177,22 +213,24 @@ class Gates(BaseActivity):
     @activity.defn(name="data_quality_gate")
     async def data_quality_gate(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
-        Data quality gate activity. for each selected filter,
+        Apply data quality filters to incoming data.
-        extracts filtered data, discards or keeps filtered data
+
-        based on the filter.
+        This activity applies configurable quality filters to validate incoming
+        data. It supports multiple filter types and provides comprehensive
+        error reporting for quality issues.
 
         Args:
-            input_data (dict[str, Any]): The data to validate. Contains:
+            input_data (dict[str, Any]): Activity input parameters.
-                filters (dict[str, str]): The filters to apply. In format:
+                Required fields:
-                    {filter_name: policy}.
+                - data (dict[str, Any]): Data to validate
-                    filter_name: The name of the filter.
+                - filters (dict[str, str]): Filter configuration
-                    policy: The policy to apply. Can be "DISCARD" or "KEEP".
+                - model_tags (dict[str, Any]): Tag-specific validation rules
-                data (dict[str, Any]): The data to validate.
-                model_tags (dict[str, Any]): The tags of the model.
-                    And it's respective configuration.
 
         Returns:
-            dict[str, Any]: The data validated.
+            dict[str, Any]: Filtered data that passes quality validation
+
+        Raises:
+            Exception: If quality validation fails
         """
 
         metadata = input_data['metadata']

scouter/activities/mongodb.py

@@ -15,13 +15,20 @@ with workflow.unsafe.imports_passed_through():
 
 def clear_mongo_id(docs: list) -> list:
     """
-    Remove the MongoDB internal `_id` field from the document.
+    Remove MongoDB internal `_id` fields from documents.
+
+    This utility function recursively removes the MongoDB `_id` field from
+    documents and nested structures. It's used to clean data before
+    processing or export operations.
 
     Args:
-        docs (list): The document to clear.
+        docs (list): List of documents to clean
 
     Returns:
-        list: The documents without the `_id` field.
+        list: Documents with `_id` fields removed
+
+    Note:
+        This function modifies the input list in-place and returns the same reference
     """
     for doc in docs:
         if isinstance(doc, list):
@@ -41,9 +48,35 @@ def clear_mongo_id(docs: list) -> list:
 
 
 class MongoDB(BaseActivity):
+    """
+    MongoDB operations for data retrieval and storage.
+
+    This class provides MongoDB connectivity and operations for the Scouter system.
+    It handles:
+    - Connection management with automatic reconnection
+    - Data retrieval with timestamp-based filtering
+    - Document cleaning and preprocessing
+    - Error handling and notification integration
+
+    The class implements Temporal activities for MongoDB operations, enabling
+    distributed data processing with fault tolerance and monitoring.
+    """
+
     def __init__(self, connection_string: str, database_name: str,
                  logger: Logger,
                  notification_handler: NotificationHandler):
+        """
+        Initialize MongoDB connection and services.
+
+        Args:
+            connection_string (str): MongoDB connection URI string
+            database_name (str): Name of the target database
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+
+        Raises:
+            ConnectionError: If MongoDB connection fails
+        """
         self.connection_string = connection_string
         self.database_name = database_name
 
@@ -63,7 +96,10 @@ class MongoDB(BaseActivity):
 
     def shutdown(self):
         """
-        Close the MongoDB client connection.
+        Gracefully close MongoDB client connection.
+
+        This method ensures proper cleanup of MongoDB connections to prevent
+        connection leaks and ensure graceful application termination.
         """
         try:
             if self.client:
@@ -75,14 +111,34 @@ class MongoDB(BaseActivity):
 
     def __del__(self):
         """
-        Destructor to ensure MongoDB client is closed when the object is deleted.
+        Destructor to ensure MongoDB client is closed.
+
+        This destructor ensures that MongoDB connections are properly closed
+        when the object is garbage collected, preventing resource leaks.
         """
         self.shutdown()
 
     @activity.defn(name="load_latest_data")
     async def load_latest_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
-        Loads the latest data from MongoDB.
+        Load the latest data from MongoDB collection since a specified timestamp.
+
+        This activity retrieves data from a MongoDB collection, optionally
+        filtering by timestamp to enable incremental data processing. It
+        handles connection management and provides comprehensive error reporting.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - collection_name (str): Name of the MongoDB collection
+                - last_data_timestamp (str | None): Last processed timestamp for filtering
+
+        Returns:
+            dict[str, Any]: Retrieved data, or empty dict if no data found
+
+        Raises:
+            Exception: If MongoDB operation fails
         """
         metadata = input_data['metadata']
         collection_name = input_data['collection_name']

scouter/activities/redis.py

@@ -14,17 +14,58 @@ with workflow.unsafe.imports_passed_through():
 
 
 class Redis(RedisBase):
+    """
+    Redis operations for data caching and temporary storage.
+
+    This class extends the base Redis functionality to provide specialized
+    operations for the Scouter system, including:
+    - Data timestamp management for incremental processing
+    - Temporary data storage with configurable TTL
+    - Data grouping and holding for batch processing
+    - Error handling and notification integration
+
+    The class implements Temporal activities for Redis operations, enabling
+    distributed data processing with fault tolerance and monitoring.
+    """
+
     def __init__(self, host: str, port: int,
                  username: str, password: str,
                  logger: Logger, notification_handler: NotificationHandler):
+        """
+        Initialize Redis connection and services.
 
+        Args:
+            host (str): Redis server hostname or IP address
+            port (int): Redis server port number
+            username (str): Redis authentication username
+            password (str): Redis authentication password
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+        """
         RedisBase.__init__(self, host, port, username,
                            password, logger, notification_handler)
 
     @activity.defn(name="get_last_data_timestamp")
     async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
         """
-        Gets the last data timestamp from redis.
+        Retrieve the last processed data timestamp from Redis.
+
+        This activity retrieves the timestamp of the last successfully processed
+        data point for a specific workflow and schedule combination. It's used
+        for incremental data processing to avoid reprocessing the same data.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - workflow_name (str): Name of the workflow
+                - schedule_name (str): Name of the data collection schedule
+
+        Returns:
+            str | None: Last processed timestamp string, or None if no previous data exists
+
+        Raises:
+            Exception: If Redis operation fails
         """
         metadata = input_data['metadata']
         key = f"last_data_timestamp:{input_data['workflow_name']}:{input_data['schedule_name']}"
@@ -55,9 +96,27 @@ class Redis(RedisBase):
         return data_hold
 
     @activity.defn(name="put_last_data_timestamp")
-    async def put_last_data_timestamp(self, input_data: dict[str, Any]):
+    async def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
         """
-        Puts the last data timestamp into redis.
+        Store the last processed data timestamp in Redis.
+
+        This activity stores the timestamp of the most recent data point that
+        has been successfully processed. The timestamp is used for incremental
+        data loading in subsequent workflow executions.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - data (dict[str, Any]): Processed data to extract timestamp from
+                - workflow_name (str): Name of the workflow
+                - schedule_name (str): Name of the data collection schedule
+
+        Returns:
+            str | None: The timestamp that was stored, or None if no data was processed
+
+        Raises:
+            Exception: If Redis operation fails
         """
         metadata = input_data['metadata']
         key = f"last_data_timestamp:{input_data['workflow_name']}:{input_data['schedule_name']}"
@@ -96,18 +155,30 @@ class Redis(RedisBase):
         return last_data_timestamp
 
     @activity.defn(name="group_and_hold_data")
-    async def group_and_hold_data(self, input_data: dict[str, Any]):
+    async def group_and_hold_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
-        Groups and holds data in redis. Keep a copy of the most recent
+        Group data by tags and store temporarily in Redis with TTL.
-        received data for a given pipeline and schedule. This activity updates
+
-        the data in redis and return the full keeped data.
+        This activity organizes processed data by tag names and stores it in Redis
+        with a configurable retention period. The data is grouped to enable
+        efficient batch processing and export operations.
 
         Args:
-            input_data (dict[str, Any]): The data to group and hold.
+            input_data (dict[str, Any]): Activity input parameters.
-                workflow_name (str): The name of the workflow.
+                Required fields:
-                schedule_name (str): The name of the schedule.
+                - metadata (dict[str, Any]): Workflow execution metadata
-                data (dict[str, Any]): The data to group and hold.
+                - schedule_name (str): Name of the data collection schedule
-                retention_time (int): The retention time for data in redis in seconds.
+                - workflow_name (str): Name of the workflow
+                - data (dict[str, Any]): Data to group and store
+                - model_id (str): Unique model identifier
+                - model_tags (dict[str, Any]): Tag configuration
+                - retention_time (int): Data retention period in seconds
+
+        Returns:
+            dict[str, Any]: Grouped data organized by tag names
+
+        Raises:
+            Exception: If Redis operation fails
         """
         metadata = input_data['metadata']
 

scouter/metrics.py

@@ -1,19 +1,23 @@
 from prometheus_client import Gauge, Counter
 
+# Application health and status metrics
 APP_UP = Gauge(
     "app_up",
     "Indicates if the application is running (1) or shutting down (0)",
     ["pod_id"],
 )
 
+# Core labels for consistent metric labeling
 CORE_LABELS = ["pod_id", "model_name", "pipeline_name"]
 
+# Data processing metrics
 LABORIOUS_DATA_WRITTEN_COUNT = Counter(
     "scouter_laborious_data_written_count",
     "Number of writings to the database table laborious_data",
     CORE_LABELS,
 )
 
+# Tag monitoring metrics
 TAG_CHANGES_MONITOR = Gauge(
     "scouter_tag_changes_monitor",
     "Current value change of each tag",

scouter/utils/connectors_config.py

@@ -1,7 +1,21 @@
 from os import getenv
+from typing import Any
 
 
-def build_postgres_config():
+def build_postgres_config() -> dict[str, Any]:
+    """
+    Build PostgreSQL connection configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: PostgreSQL configuration dictionary with keys:
+            - host: Database hostname (default: localhost)
+            - port: Database port (default: 5432)
+            - user: Database username (default: sientia)
+            - password: Database password (default: sientia)
+            - dbname: Database name (default: sientia)
+            - min_connections: Minimum connection pool size (default: 5)
+            - max_connections: Maximum connection pool size (default: 20)
+    """
     return {
         'host': getenv('POSTGRES_HOST', 'localhost'),
         'port': int(getenv('POSTGRES_PORT', '5432')),
@@ -13,7 +27,16 @@ def build_postgres_config():
     }
 
 
-def build_kafka_config():
+def build_kafka_config() -> dict[str, Any]:
+    """
+    Build Kafka configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: Kafka configuration dictionary with keys:
+            - bootstrap_servers: Kafka broker addresses (default: localhost:9092)
+            - polling_time: Consumer polling interval in milliseconds (default: 1000)
+            - group_id: Consumer group identifier (default: scouter-group)
+    """
     return {
         'bootstrap_servers': getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
         'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')),
@@ -21,7 +44,17 @@ def build_kafka_config():
     }
 
 
-def build_redis_config():
+def build_redis_config() -> dict[str, Any]:
+    """
+    Build Redis connection configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: Redis configuration dictionary with keys:
+            - host: Redis server hostname (default: localhost)
+            - port: Redis server port (default: 6379)
+            - username: Redis authentication username (default: None)
+            - password: Redis authentication password (default: None)
+    """
     return {
         'host': getenv('REDIS_HOST', 'localhost'),
         'port': int(getenv('REDIS_PORT', '6379')),
@@ -30,7 +63,15 @@ def build_redis_config():
     }
 
 
-def build_mongodb_config():
+def build_mongodb_config() -> dict[str, Any]:
+    """
+    Build MongoDB connection configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: MongoDB configuration dictionary with keys:
+            - connection_string: Complete MongoDB connection URI
+            - database_name: Target database name (default: sientia)
+    """
     username = getenv('MONGODB_USERNAME', 'sientia')
     password = getenv('MONGODB_PASSWORD', 'sientia')
     uri = getenv('MONGODB_URL', 'localhost:27017')
@@ -42,7 +83,15 @@ def build_mongodb_config():
     }
 
 
-def build_druid_config():
+def build_druid_config() -> dict[str, Any]:
+    """
+    Build Apache Druid connection configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: Druid configuration dictionary with keys:
+            - host: Druid server hostname (default: localhost)
+            - port: Druid server port (default: 8082)
+    """
     return {
         'host': getenv('DRUID_HOST', 'localhost'),
         'port': int(getenv('DRUID_PORT', '8082')),

scouter/utils/quality/filters.py

@@ -5,14 +5,21 @@ from typing import Any
 
 def check_data_range(value: float | int | None, val_range: list) -> bool:
     """
-    Check if a value is out of a given range.
+    Check if a value falls outside the specified range.
+
+    This function validates if a numeric value is within the acceptable range
+    defined by the minimum and maximum bounds. It handles edge cases including
+    None values and NaN values.
 
     Args:
-        value (float | int | None): The value to check.
+        value (float | int | None): The numeric value to validate
-        val_range (list): The range to check against.
+        val_range (list): List containing [min_value, max_value] bounds
 
     Returns:
-        bool: True if the value is out of the range, False otherwise.
+        bool: True if value is outside the range, False if within range
+
+    Note:
+        None and NaN values are considered out of range (return True)
     """
     if value is None or np.isnan(value):
         return True
@@ -23,32 +30,46 @@ def check_data_range(value: float | int | None, val_range: list) -> bool:
     return value < bottom or value > up
 
 
-def out_of_bounds_filter(df: DataFrame, model_tags: dict[str, Any]):
+def out_of_bounds_filter(df: DataFrame, model_tags: dict[str, Any]) -> DataFrame:
     """
-    Filter out rows where the value is out of the range.
+    Filter DataFrame rows where values are outside configured ranges.
+
+    This function applies range validation to each row in the DataFrame based
+    on tag-specific configuration. Rows with values outside the configured
+    ranges are filtered out.
 
     Args:
-        df (DataFrame): The DataFrame to filter.
+        df (DataFrame): DataFrame containing sensor data with 'name' and 'value' columns
-        model_tags (dict[str, Any]): The model tags. Contains
+        model_tags (dict[str, Any]): Tag configuration containing data_range for each tag.
-        the data_range for each tag. If the tag does not have a data_range,
+            If a tag doesn't have data_range, it's considered to have infinite bounds.
-        it will be considered as (-inf, inf).
 
     Returns:
-        DataFrame: The filtered DataFrame.
+        DataFrame: Filtered DataFrame with out-of-bounds values removed
+
+    Note:
+        Tags without data_range configuration are treated as having infinite bounds
     """
     return df[df.apply(lambda x: check_data_range(
         x['value'], model_tags[x['name']].get('data_range', (-np.inf, np.inf))),
         axis=1)]
 
 
-def null_values_filter(df: DataFrame, _model_tags: dict[str, Any]):
+def null_values_filter(df: DataFrame, _model_tags: dict[str, Any]) -> DataFrame:
     """
-    Filter out rows where the value is null.
+    Filter DataFrame rows containing null values.
+
+    This function removes rows where the 'value' column contains null values.
+    It's used for data quality filtering to ensure only complete data records
+    are processed.
 
     Args:
-        df (DataFrame): The DataFrame to filter.
+        df (DataFrame): DataFrame containing sensor data with 'value' column
+        _model_tags (dict[str, Any]): Tag configuration (unused in this filter)
 
     Returns:
-        DataFrame: The filtered DataFrame.
+        DataFrame: Filtered DataFrame with null values removed
+
+    Note:
+        The _model_tags parameter is included for interface consistency but not used
     """
     return df[df['value'].isnull()]

scouter/worker/worker.py

@@ -21,11 +21,32 @@ with workflow.unsafe.imports_passed_through():
         build_mongodb_config
     )
 
+# Environment configuration
 POD_ID = os.getenv("HOSTNAME", "localhost")
 SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', "9091"))
 
 
 async def main():
+    """
+    Main entry point for the Scouter Temporal worker.
+
+    This function initializes and starts all required services:
+    - Prometheus metrics server
+    - Notification handler for MongoDB
+    - Activity implementations for data processing
+    - Temporal client and workers
+    - Multiple task queues for different workflow types
+
+    The worker supports two main task queues:
+    - scouter-queue: Main data processing workflows
+    - fake_data-queue: Test data generation workflows
+
+    Returns:
+        None
+
+    Raises:
+        SystemExit: If worker initialization or execution fails
+    """
     host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
 
     logger = get_logger(__name__)
@@ -148,6 +169,19 @@ async def main():
 
 
 def start_prometheus_server():
+    """
+    Start the Prometheus metrics HTTP server.
+
+    This function initializes the Prometheus metrics server on the configured
+    port and sets the application health status. It's essential for
+    monitoring and observability of the Scouter system.
+
+    Returns:
+        None
+
+    Raises:
+        SystemExit: If metrics server fails to start
+    """
     try:
         port = int(os.getenv("HTTP_METRICS_PORT", 9090))
         start_http_server(port)

scouter/workflow/fake_data.py

@@ -9,14 +9,43 @@ with workflow.unsafe.imports_passed_through():
 
 @workflow.defn(name="fake_data")
 class FakeData:
+    """
+    Test data generation workflow for development and testing purposes.
+
+    This workflow generates synthetic industrial sensor data and publishes it to
+    Kafka topics. It's designed for:
+    - Development and testing of data processing pipelines
+    - Load testing of downstream systems
+    - Demonstration of data flow patterns
+    - Validation of data quality filters and aggregation functions
+
+    The generated data simulates realistic industrial sensor readings with
+    configurable message counts and topic routing.
+    """
+
     @workflow.run
     async def run(self, workflow_input: Dict[str, Any]) -> str:
         """
-        Generates random data and sends it to a Kafka topic.
+        Execute the fake data generation workflow.
+
+        This method generates synthetic sensor data and publishes it to the specified
+        Kafka topic. The data includes realistic industrial sensor readings with
+        configurable parameters for testing and development purposes.
 
         Args:
-            workflow_input (dict[str, Any]): The input data containing:
+            workflow_input (dict[str, Any]): Workflow configuration parameters.
-                topic (str): The Kafka topic to send data to
+                Required fields:
+                - topic (str): Kafka topic name for data publication
+                - metadata (dict[str, Any], optional): Workflow execution metadata
+                - num_messages (int, optional): Number of messages to generate.
+                  Defaults to random count between 1 and available sensor tags.
+
+        Returns:
+            str: Success confirmation message
+
+        Raises:
+            WorkflowExecutionError: If workflow execution fails
+            ActivityExecutionError: If data generation or Kafka publishing fails
         """
         await workflow.execute_activity_method(
             Faker.generate_and_send_data,

scouter/workflow/scouter.py

@@ -9,25 +9,55 @@ with workflow.unsafe.imports_passed_through():
 
 @workflow.defn(name="scouter")
 class Scouter:
-    @workflow.run
-    async def run(self, input_data: dict[str, Any]):
     """
-        Scouter workflow. Loads data from kafka and sends it to the core_scouter
+    Main Scouter workflow that orchestrates data ingestion and processing.
-        workflow.
+
+    This workflow serves as the entry point for data processing pipelines. It loads
+    data from MongoDB collections, manages data timestamps for incremental processing,
+    and delegates the actual data processing to the CoreScouter workflow.
+
+    The workflow implements a robust data ingestion pattern with:
+    - Incremental data loading based on last processed timestamp
+    - Automatic timestamp management for data continuity
+    - Error handling and retry policies
+    - Child workflow orchestration for data processing
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]) -> None:
+        """
+        Execute the main Scouter workflow.
+
+        This method orchestrates the complete data ingestion process:
+        1. Retrieves the last processed timestamp from Redis
+        2. Loads new data from MongoDB since the last timestamp
+        3. Updates the last processed timestamp
+        4. Delegates data processing to the CoreScouter workflow
 
         Args:
-            input_data (dict[str, Any]): The data to process. Contains:
+            input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
-                topic (str): The topic to load data from.
+                Required fields:
-                schedule_name (str): The name of the schedule.
+                - topic (str): The Kafka topic name for data source identification
-                model_name (str): The name of the model.
+                - schedule_name (str): Unique identifier for the data collection schedule
-                model_id (str): The id of the model.
+                - model_name (str): Name of the data model being processed
-                trigger_laborious (bool): Whether to trigger laborious.
+                - model_id (str): Unique identifier for the data model
-                filters (dict[str, str]): The filters to apply.
+                - trigger_laborious (bool): Flag to enable intensive data processing
-                schema (str): The schema of the table to export data to.
+                - filters (dict[str, str]): Data quality filters configuration
-                table_name (str): The name of the table to export data to.
+                - schema (str): Target database schema for data export
-                retention_time (int): The retention time for data in redis in seconds.
+                - table_name (str): Target table name for data export
-                model_tags (dict[str, Any]): The tags of the model.
+                - retention_time (int): Data retention period in Redis (seconds)
-                    And it's respective configuration.
+                - model_tags (dict[str, Any]): Tag-specific configuration including:
+                    - data_range: [min, max] values for data validation
+                    - aggr_function: Aggregation method (avg, mdn, max, min, lts)
+                    - frequency: Data collection frequency in milliseconds
+                    - topics: List of Kafka topics for data routing
+
+        Returns:
+            None: This workflow doesn't return data, it orchestrates data processing
+
+        Raises:
+            WorkflowExecutionError: If workflow execution fails
+            ActivityExecutionError: If any activity fails after retry attempts
         """
 
         input_data['workflow_name'] = 'scouter'

scouter/workflow/sub_workflows/core_scouter.py

@@ -10,27 +10,54 @@ with workflow.unsafe.imports_passed_through():
 
 @workflow.defn(name="core_scouter")
 class CoreScouter:
-    @workflow.run
-    async def run(self, input_data: dict[str, Any]):
     """
-        Core scouter workflow. Passes data through data_quality_gate,
+    Core data processing workflow that handles data quality, aggregation, and export.
-        group_and_hold_data, and then asynchronously exports data to postgres
+
-        using export_data_to_postgres and in the future will trigger_laborious
+    This workflow implements the core data processing pipeline for industrial data:
-        if needed.
+    - Data quality validation and filtering
+    - Time-series data aggregation using configurable functions
+    - Data grouping and temporary storage in Redis
+    - Asynchronous export to PostgreSQL for persistent storage
+    - Metrics collection and monitoring
+
+    The workflow is designed for high-throughput data processing with configurable
+    quality gates and aggregation strategies.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]) -> None:
+        """
+        Execute the core data processing workflow.
+
+        This method processes industrial time-series data through a series of stages:
+        1. Data Quality Gate: Applies configurable filters for data validation
+        2. Data Aggregation: Groups and aggregates data using specified functions
+        3. Data Grouping: Organizes data by tags and applies retention policies
+        4. Data Export: Persists processed data to PostgreSQL
+        5. Metrics Collection: Records processing metrics for monitoring
 
         Args:
-            input_data (dict[str, Any]): The data to process. Contains:
+            input_data (dict[str, Any]): Complete workflow configuration and data.
-                metadata (dict[str, Any]): The metadata of the workflow.
+                Required fields:
-                workflow_name (str): The name of the workflow.
+                - metadata (dict[str, Any]): Workflow execution metadata
-                schedule_name (str): The name of the schedule.
+                - workflow_name (str): Name of the parent workflow
-                model_name (str): The name of the model.
+                - schedule_name (str): Data collection schedule identifier
-                model_id (str): The id of the model.
+                - model_name (str): Data model name
-                data (dict[str, Any]): The data to process.
+                - model_id (str): Unique model identifier
-                trigger_laborious (bool): Whether to trigger laborious.
+                - data (dict[str, Any]): Raw time-series data to process
-                filters (dict[str, str]): The filters to apply.
+                - trigger_laborious (bool): Enable intensive processing mode
-                schema (str): The schema of the table to export data to.
+                - filters (dict[str, str]): Data quality filter configurations
-                table_name (str): The name of the table to export data to.
+                - schema (str): Target database schema
-                retention_time (int): The retention time for data in redis in seconds.
+                - table_name (str): Target database table
+                - retention_time (int): Redis data retention period (seconds)
+                - model_tags (dict[str, Any]): Tag-specific processing rules
+
+        Returns:
+            None: This workflow processes data but doesn't return results
+
+        Raises:
+            WorkflowExecutionError: If workflow execution fails
+            ActivityExecutionError: If any activity fails after retry attempts
         """
 
         metadata = input_data['metadata']

tests/activities/test_activities.py

@@ -133,5 +133,5 @@ def test_shutdown(mock_mongodb_close, mock_postgres_close, _mock_mongodb_init,
 
     activities.shutdown()
 
-    mock_postgres_close.assert_called_once()
+    mock_postgres_close.assert_called()
-    mock_mongodb_close.assert_called_once()
+    mock_mongodb_close.assert_called()

values.yaml

@@ -3,7 +3,7 @@
 # Declare variables to be passed into your templates.
 
 # This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
-replicaCount: 3
+replicaCount: 1
 
 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.4.4"
+  tag: "0.4.5"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -150,7 +150,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal"
+    value: "main"
   - name: PYTHON_APP
     value: "scouter.worker.worker"