diff --git a/.env b/.env new file mode 100644 index 0000000..af9d68d --- /dev/null +++ b/.env @@ -0,0 +1,4 @@ +# === Simulator Git Repo === +# Use SSH format because the Dockerfile uses SSH to clone +SIMULATOR_GIT_REPO=git@github.com:Aignosi/sientia-dataops-opc_simulator.git +SIMULATOR_GIT_BRANCH=main diff --git a/.gitignore b/.gitignore index 42893e0..a34260d 100644 --- a/.gitignore +++ b/.gitignore @@ -12,7 +12,7 @@ docker-compose.override.yml **/deploy/*.yaml scouter/.file_versions/ scouter/pipelines/**/triggers.yaml - +**/postgres_data/** # Ignorar arquivos e diretórios de cache do Python __pycache__/ *.pyc diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..532cee8 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,82 @@ +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 + 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 + + simulator: + build: + context: . + dockerfile: simulator/Dockerfile + args: + GIT_REPO: ${SIMULATOR_GIT_REPO} + GIT_BRANCH: ${SIMULATOR_GIT_BRANCH} + container_name: simulator + ports: + - "4840:4840" + depends_on: + - kafka + networks: + - sientia-network + env_file: + - .env + + +networks: + sientia-network: + driver: bridge + +volumes: + postgres_data: + driver: local \ No newline at end of file diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index af7268a..c7d1821 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -50,5 +50,5 @@ class Activities(Postgres, MLFlow, Gates, OPC): notification_handler=notification_handler) @activity.defn(name="prepare_activity") - async def prepare_activity(self, schedule_name: str, model_name: str, model_id: str): - await super().prepare_activity(schedule_name, model_name, model_id) + def prepare_activity(self, input_data: dict[str, Any]): + super().prepare_activity(input_data) diff --git a/laborious/activities/base.py b/laborious/activities/base.py index 9f89e5e..2742457 100644 --- a/laborious/activities/base.py +++ b/laborious/activities/base.py @@ -1,6 +1,6 @@ +from logging import Logger from temporalio import activity from sientia_do.notifications.handlers import NotificationHandler -from logging import Logger class BaseActivity: @@ -8,17 +8,18 @@ class BaseActivity: self.logger = logger self.notification_handler = notification_handler - def prepare_activity(self, schedule_name: str, - model_name: str, - model_id: str): + @activity.defn(name="prepare_activity") + def prepare_activity(self, input_data: dict[str, Any]): """ Prepare the activity for the notification handler. Args: + workflow_name (str): The name of the workflow. schedule_name (str): The name of the schedule. model_name (str): The name of the model. model_id (str): The id of the model. """ - self.notification_handler.base_notification.schedule_name = schedule_name - self.notification_handler.base_notification.model_name = model_name - self.notification_handler.base_notification.model_id = model_id + self.notification_handler.base_notification.pipeline_name = input_data['workflow_name'] + self.notification_handler.base_notification.schedule_name = input_data['schedule_name'] + self.notification_handler.base_notification.model_name = input_data['model_name'] + self.notification_handler.base_notification.model_id = input_data['model_id'] diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py new file mode 100644 index 0000000..8f98624 --- /dev/null +++ b/laborious/worker/worker.py @@ -0,0 +1,105 @@ +from temporalio import workflow, client +from temporalio.worker import Worker + +with workflow.unsafe.imports_passed_through(): + from laborious.workflows.predictions_batch import PredictionsBatch + from laborious.activities.activities import Activities + import os + import logging + from sientia_do.notifications.handlers import NotificationHandler + + +async def main(): + host = os.getenv('TEMPORAL_HOST', 'localhost:7233') + logger = logging.getLogger(__name__) + stream_handler = logging.StreamHandler() + stream_handler.setLevel( + os.getenv('LOG_LEVEL', 'INFO').upper() + ) + stream_handler.setFormatter( + logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + ) + + logger.addHandler(stream_handler) + + notification_handler = NotificationHandler( + servers=os.getenv('NOTIFICATION_SERVERS', 'http://localhost:29092'), + logger=logger, + project_name=os.getenv('PROJECT_NAME', 'laborious'), + pipeline_name='-', + trigger_name='-', + model_name='-', + model='-' + ) + + postgres_config = { + 'host': os.getenv('POSTGRES_HOST', 'localhost'), + 'port': int(os.getenv('POSTGRES_PORT', '5432')), + 'user': os.getenv('POSTGRES_USER', 'postgres'), + 'password': os.getenv('POSTGRES_PASSWORD', 'postgres'), + 'dbname': os.getenv('POSTGRES_DBNAME', 'postgres'), + 'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '5')), + 'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '20')) + } + + mlflow_config = { + 'host': os.getenv('MLFLOW_HOST', 'localhost'), + 'port': int(os.getenv('MLFLOW_PORT', '5000')), + 'username': os.getenv('MLFLOW_USERNAME', 'aignosi'), + 'password': os.getenv('MLFLOW_PASSWORD', 'aignosi') + } + + opc_config = { + 'name': os.getenv('OPC_NAME', 'opc'), + 'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'), + 'server_uri': os.getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), + 'cert_path': os.getenv('OPC_CERT_PATH', None), + 'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', None), + 'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', None) + } + + activities = Activities( + postgres_config=postgres_config, + mlflow_config=mlflow_config, + opc_config=opc_config, + logger=logger, + notification_handler=notification_handler + ) + + temporal_client = await client.Client.connect(target_host=host) + workers = [ + Worker( + temporal_client, + task_queue='predictions', + workflows=[PredictionsBatch], + activities=[ + # Base + activities.prepare_activity, + # MLFlow + activities.request_predict, + activities.request_transform, + # Gates + activities.input_gate, + activities.mlflow_response_gate, + activities.mlflow_content_gate, + 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, + activities.export_data_to_postgres + ] + ) + ] + + for w in workers: + await w.run() + +if __name__ == '__main__': + import asyncio + asyncio.run(main()) diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 6c137da..55eff80 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -15,7 +15,8 @@ class PredictionsBatch(): { 'schedule_name': input_data['schedule_name'], 'model_name': input_data['model_name'], - 'model_id': input_data['model_id'] + 'model_id': input_data['model_id'], + 'workflow_name': 'predictions_batch' } ) diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 553c9f8..7d6d2ec 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -70,6 +70,7 @@ class FormatAndExportPrediction(): } ) + # write to opc opc_holder = workflow.execute_activity_method( Activities.write_opc_data, { diff --git a/simulator/Dockerfile b/simulator/Dockerfile new file mode 100644 index 0000000..d467676 --- /dev/null +++ b/simulator/Dockerfile @@ -0,0 +1,30 @@ +# syntax=docker/dockerfile:1.4 + +FROM python:3.11-slim + +# Enable use of SSH agent/socket +# This line enables SSH during build +# (don't forget the syntax header above) +RUN apt-get update && apt-get install -y git openssh-client && rm -rf /var/lib/apt/lists/* + +# Use build-time SSH mount for Git clone +# The SSH key will NOT remain in the image +# IMPORTANT: this block requires BuildKit +# and the --ssh flag during docker build + +# SSH config to skip host key check (safe in CI/local dev) +RUN mkdir -p /root/.ssh && echo "StrictHostKeyChecking no" > /root/.ssh/config + +WORKDIR /app + +# Clone using SSH +ARG GIT_REPO +ARG GIT_BRANCH=main + +# Mount SSH key just for this RUN +RUN --mount=type=ssh git clone --branch ${GIT_BRANCH} ${GIT_REPO} . + +# Install requirements if exists +RUN if [ -f requirements.txt ]; then pip install --no-cache-dir -r requirements.txt; fi + +CMD ["python", "server.py"] diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py new file mode 100644 index 0000000..3f7e36a --- /dev/null +++ b/simulator/redis-feeder.py @@ -0,0 +1,55 @@ +import redis +import json +import os + +# Redis connection settings +redis_host = "localhost" +redis_port = 6379 + +# Connect to Redis +r = redis.Redis(host=redis_host, port=redis_port, + decode_responses=True, username='default', password='bdnZOpcyiL') + +# Define the key pattern to target +pattern = "slot:opc_tags:*" + +# Step 1: Find and delete matching keys +print("🔍 Searching for keys matching:", pattern) +for key in r.scan_iter(match=pattern): + r.delete(key) + print(f"❌ Deleted: {key}") + +# Step 2: Insert new data +# Example new OPC tag data +new_data = { + "slot:opc_tags:1": { + "server1": { + "name": "server1", + "url": "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840", + "server_uri": "http://opcua-server.simulator", + "tags": { + 'ns=2;i=2': { + 'tag_name': 'Counter', + 'frequency': 1000, + 'topics': ['opcua', 'counter'], + }, + 'ns=2;i=3': { + 'tag_name': 'Rollout', + 'frequency': 1000, + "topics": ['opcua', 'rollout'], + }, + 'ns=2;i=4': { + 'tag_name': 'Square', + 'frequency': 1000, + "topics": ['opcua'], + }, + } + } + } +} + +for key, val in new_data.items(): + r.set(key, json.dumps(val)) + print(f"✅ Set: {key} -> {val}") + +print("🚀 OPC tag keys replaced successfully.")