Merge pull request #1 from Aignosi/SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas

Sientiapde 1030 implementar os workflows mapeados utilizando as workers e activities apropriadas

.github/workflows/quality-gate.yml

@@ -0,0 +1,109 @@
+name: Quality gate
+
+on:
+  push:
+    branches:
+      - '**'
+  pull_request:
+    branches:
+      - '**'
+    types: [ opened, synchronize, reopened ]
+
+jobs:
+  sonar:
+    name: SonarQube Analysis
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+
+    steps:
+      - name: ⬇️ Checkout Code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Generate App Token
+        id: generate-app-token
+        uses: actions/create-github-app-token@v1
+        with:
+          app-id: ${{ secrets.APP_ID }}
+          private-key: ${{ secrets.APP_PRIVATE_KEY }}
+          owner: 'Aignosi'
+          repositories: 'sientia-dataops-library'
+
+      - name: Prepare requirements.txt
+        id: prepare-requirements
+        run: |
+          sed -e "s|git+ssh://git@github.com/|git+https://github.com/|g" \
+              -e "s|git@github.com:|git+https://github.com/|g" \
+              requirements.txt > requirements_prepared.txt
+          echo "PROCESSED_REQUIREMENTS_FILE=requirements_prepared.txt" >> $GITHUB_OUTPUT
+
+      - name: Configure Git to use App Token
+        env:
+          GH_APP_TOKEN: ${{ steps.generate-app-token.outputs.token }}
+        run: |
+          git config --global url."https://oauth2:${GH_APP_TOKEN}@github.com/".insteadOf "https://github.com/"
+
+      - name: 🔧 Setup Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+
+      - name: 🗄️ Cache Python dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles(steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE) }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: 📦 Install Dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }}
+          pip install pytest pytest-cov pytest-asyncio
+      
+      - name: ⬇️ Setup Node.js 18
+        uses: actions/setup-node@v4
+        with:
+          node-version: 18
+
+      - name: 📥 Setup SonarScanner
+        uses: warchant/setup-sonar-scanner@v7
+
+      - name: 🧪 Run Tests with Pytest
+        run: |
+          set +e 
+          pytest tests --junitxml=pytest.xml --cov=orchestrator --cov-report=xml --cov-report=term
+          PYTEST_EXIT_CODE=$?
+          set -e
+
+          if [ $PYTEST_EXIT_CODE -eq 0 ]; then
+            echo "Pytest executado com sucesso."
+          elif [ $PYTEST_EXIT_CODE -eq 5 ]; then
+            echo "Pytest finalizado com código 5 (Nenhum teste encontrado). Tratando como sucesso para este workflow."
+            exit 0 
+          else
+            echo "Pytest falhou com código de saída $PYTEST_EXIT_CODE."
+            exit $PYTEST_EXIT_CODE 
+          fi
+
+      - name: 📊 Run SonarQube Analysis
+        env:
+          SONAR_PROJECT_KEY: ${{ secrets.SONAR_PROJECT_KEY }}
+          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} # Este é o token do SonarQube
+        run: |
+          sonar-scanner \
+            -Dsonar.projectKey=$SONAR_PROJECT_KEY \
+            -Dsonar.sources=orchestrator \
+            -Dsonar.tests=tests \
+            -Dsonar.python.coverage.reportPaths=coverage.xml \
+            -Dsonar.python.xunit.reportPath=pytest.xml \
+            -Dsonar.host.url=$SONAR_HOST_URL \
+            -Dsonar.token=$SONAR_TOKEN \
+            -Dsonar.python.version=3.11 \
+            -Dsonar.projectVersion=1.0.0 \
+            -Dsonar.coverage.exclusions=orchestrator/worker/worker.py

.gitignore

@@ -1,174 +1,44 @@
-# Byte-compiled / optimized / DLL files
+# Ignorar volumes do Docker
+docker-compose.override.yml
+**/db_data/
+**/kafka-volume/
+**/zookeeper-volume/
+**/mage_data/
+**/minio_data/
+**/venv/
+**/certs/*.pem
+**/certs/*.der
+**/certs/*.csr
+**/deploy/*.yaml
+scouter/.file_versions/
+scouter/pipelines/**/triggers.yaml
+**/postgres_data/**
+**/couchbase_data/**
+**/redis_data/**
+# Ignorar arquivos e diretórios de cache do Python
 __pycache__/
-*.py[cod]
+*.pyc
-*$py.class
+*.pyo
+*.pyd
 
-# C extensions
+# Ignorar logs
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-cover/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
 *.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
 
-# Flask stuff:
+# Ignorar arquivos de configuração locais
-instance/
+.vscode/
-.webassets-cache
+.pytest_cache/
+.idea/
+*.swp
 
-# Scrapy stuff:
+# Ignorar arquivos temporários
-.scrapy
+*.tmp
+*.bak
+*.old
+.secret
 
-# Sphinx documentation
+# Ignorar coverage
-docs/_build/
+htmlcov/
+.coverage
 
-# PyBuilder
+# git keys
-.pybuilder/
+git_key*
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-#   For a library or package, you might want to ignore these files since the code is
-#   intended to run in multiple environments; otherwise, check them in:
-# .python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# UV
-#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-#uv.lock
-
-# poetry
-#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
-#poetry.lock
-
-# pdm
-#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
-#pdm.lock
-#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
-#   in version control.
-#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
-.pdm.toml
-.pdm-python
-.pdm-build/
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
-# pytype static type analyzer
-.pytype/
-
-# Cython debug symbols
-cython_debug/
-
-# PyCharm
-#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
-#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#  and can be added to the global gitignore or merged into this file.  For a more nuclear
-#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
-
-# Ruff stuff:
-.ruff_cache/
-
-# PyPI configuration file
-.pypirc

Dockerfile

@@ -0,0 +1,83 @@
+FROM python:3.11-bookworm
+LABEL description="Deploy Mage on ECS"
+ARG FEATURE_BRANCH
+USER root
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+
+# Definir Python 3.11 como padrão
+ENV PATH="/usr/local/bin/python3.11:$PATH"
+RUN update-alternatives --install /usr/bin/python python /usr/local/bin/python3.11 1 && \
+    update-alternatives --install /usr/bin/python3 python3 /usr/local/bin/python3.11 1 && \
+    update-alternatives --config python3 <<< '1' && \
+    update-alternatives --config python <<< '1'
+
+## System Packages
+RUN \
+  curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \
+  curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
+  apt-get -y update && \
+  ACCEPT_EULA=Y apt-get -y install --no-install-recommends \
+  # NFS dependencies
+  nfs-common \
+  # odbc dependencies
+  msodbcsql18 \
+  unixodbc-dev \
+  graphviz \
+  # postgres dependencies
+  postgresql-client \
+  # R
+  r-base && \
+  apt-get clean && \
+  rm -rf /var/lib/apt/lists/*
+
+## R Packages
+RUN \
+  R -e "install.packages('pacman', repos='http://cran.us.r-project.org')" && \
+  R -e "install.packages('renv', repos='http://cran.us.r-project.org')"
+
+## Python Packages
+RUN \
+  pip3 install --no-cache-dir sparkmagic && \
+  mkdir ~/.sparkmagic && \
+  curl https://raw.githubusercontent.com/jupyter-incubator/sparkmagic/master/sparkmagic/example_config.json > ~/.sparkmagic/config.json && \
+  sed -i 's/localhost:8998/host.docker.internal:9999/g' ~/.sparkmagic/config.json && \
+  jupyter-kernelspec install --user "$(pip3 show sparkmagic | grep Location | cut -d' ' -f2)/sparkmagic/kernels/pysparkkernel"
+
+# Mage integrations and other related packages
+RUN \
+  pip3 install --no-cache-dir "git+https://github.com/wbond/oscrypto.git@d5f3437ed24257895ae1edd9e503cfb352e635a8" && \
+  pip3 install --no-cache-dir "git+https://github.com/dremio-hub/arrow-flight-client-examples.git#egg=dremio-flight&subdirectory=python/dremio-flight" && \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/singer-python.git#egg=singer-python" && \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/dbt-mysql.git#egg=dbt-mysql" && \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/sqlglot#egg=sqlglot" && \
+  pip3 install --no-cache-dir faster-fifo && \
+  if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ]; then \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git#egg=mage-integrations&subdirectory=mage_integrations"; \
+  else \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-integrations&subdirectory=mage_integrations"; \
+  fi
+
+# Mage
+COPY ./mage_ai/server/constants.py /tmp/constants.py
+RUN if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ] ; then \
+  tag=$(tail -n 1 /tmp/constants.py) && \
+  VERSION=$(echo "$tag" | tr -d "'") && \
+  pip3 install --no-cache-dir "mage-ai[all]==$VERSION"; \
+  else \
+  pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-ai[all]"; \
+  fi
+
+## Startup Script
+COPY --chmod=0755 ./scripts/install_other_dependencies.py ./scripts/run_app.sh /app/
+ENV MAGE_DATA_DIR="/home/src/mage_data"
+ENV PYTHONPATH="${PYTHONPATH}:/home/src"
+WORKDIR /home/src
+EXPOSE 6789
+EXPOSE 7789
+
+# Copia o arquivo requirements.txt para o contêiner
+COPY requirements.txt /app/requirements.txt
+RUN pip3 install --no-cache-dir -r /app/requirements.txt
+
+
+CMD ["/bin/sh", "-c", "/app/run_app.sh"]

Makefile

@@ -0,0 +1,7 @@
+VERSION = 1.0.8
+name = sientia-laborious
+# ENVIRONMENT = production
+
+docker-hub:
+	@docker build --no-cache -t aignosi.azurecr.io/$(name):$(VERSION) .
+	@docker push aignosi.azurecr.io/$(name):$(VERSION)

README.md

@@ -1,2 +0,0 @@
-# sientia-dataops-orchestrator_temporal
-Orchestrator for SIENTIA at Temporal frameworker

docker-compose.yml

@@ -0,0 +1,133 @@
+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
+
+  
+
+  couchbase:
+    image: couchbase/server:7.2.0
+    container_name: couchbase
+    ports:
+      - "8091:8091" # Admin UI and REST API
+      - "8092:8092" # Query Service (N1QL)
+      - "8093:8093" # Index Service
+      - "8094:8094" # Search Service
+      - "11210:11210" # Data Service (KV)
+      - "18091:18091" # Analytics Service (if enabled)
+    environment:
+      CB_CLUSTER_USERNAME: sientia
+      CB_CLUSTER_PASSWORD: sientia
+      CB_CLUSTER_RAMSIZE: 256
+      CB_CLUSTER_INDEX_RAMSIZE: 256
+    volumes:
+      - ./couchbase_data:/opt/couchbase/var
+    networks:
+      - sientia-network
+    healthcheck:
+      test: ["CMD-SHELL", "curl -f http://localhost:8091/pools/default || exit 1"]
+      interval: 10s
+      timeout: 10s
+      retries: 5
+
+  redis:
+    image: redis:7-alpine # Using a lightweight Redis image
+    container_name: redis
+    ports:
+      - "6379:6379"
+    volumes:
+      - ./redis_data:/data # Persist Redis data
+    networks:
+      - sientia-network
+
+  redis-commander:
+    image: rediscommander/redis-commander:latest
+    container_name: redis-commander
+    environment:
+      REDIS_HOSTS: local:redis:6379 # Connects to the 'redis' service within the Docker network
+    ports:
+      - "8081:8081" # Access the Redis Commander UI on this port
+    depends_on:
+      - redis # Ensures Redis starts before Redis Commander
+    networks:
+      - sientia-network
+
+  kafka:
+    image: bitnami/kafka:3.7 # Using a specific Kafka version for stability
+    container_name: kafka
+    ports:
+      - "9092:9092" # For clients connecting from the host machine or outside Docker network
+    environment:
+      # KRaft (Kafka Raft without Zookeeper) settings
+      KAFKA_CFG_NODE_ID: '0'
+      KAFKA_CFG_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      # Listeners: <PROTOCOL>://<HOST/IP>:<PORT>
+      # PLAINTEXT_EXTERNAL for host access, INTERNAL for container-to-container communication
+      KAFKA_CFG_LISTENERS: 'PLAINTEXT_EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093'
+      # Advertised Listeners: How clients (including Kafka-UI) will connect
+      KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT_EXTERNAL://localhost:9092,INTERNAL://kafka:19092'
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT_EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT'
+      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: 'INTERNAL'
+      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093' # Node 0 is at kafka:9093 for controller comms
+
+      # Single node cluster settings (important for KRaft single node)
+      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: '1'
+      KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: '1'
+      KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: '1'
+      KAFKA_CFG_DEFAULT_REPLICATION_FACTOR: '1' # For auto-created topics
+      KAFKA_CFG_NUM_PARTITIONS: '1' # Default partitions for auto-created topics
+
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true' # Convenient for development
+    volumes:
+      - kafka_data:/bitnami/kafka # Bitnami Kafka data directory
+    networks:
+      - sientia-network
+    healthcheck:
+      # Checks if Kafka is ready by trying to list topics using the internal listener
+      test: ["CMD-SHELL", "/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:19092 --list > /dev/null || exit 1"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    ports:
+      - "8082:8080" # Kafka UI will be accessible on host's port 8082
+    environment:
+      KAFKA_CLUSTERS_0_NAME: sientia-local-kafka
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092 # Connects to Kafka's internal listener
+      # DYNAMIC_CONFIG_ENABLED: 'true' # Optional: To allow config changes through UI
+    depends_on:
+      kafka: # Ensures Kafka starts and is healthy before Kafka UI
+        condition: service_healthy
+    networks:
+      - sientia-network
+
+
+networks:
+  sientia-network:
+    driver: bridge
+
+volumes:
+  postgres_data:
+    driver: local
+  couchbase_data:
+    driver: local
+  redis_data:
+    driver: local
+  kafka_data:
+    driver: local

input_sample.json

@@ -0,0 +1,5 @@
+{
+    "schedule_name": "orchestrator-test",
+    "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;",
+    "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;"
+}

orchestrator/activities/activities.py

@@ -0,0 +1,52 @@
+from temporalio import activity, workflow
+from temporalio.client import Client
+
+with workflow.unsafe.imports_passed_through():
+    from orchestrator.activities.couchbase import Couchbase
+    from orchestrator.activities.temporal_manager import TemporalManager
+    from orchestrator.activities.slot_manager import SlotManager
+    from orchestrator.activities.formatters import Formatters
+    from typing import Any
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+class Activities(Couchbase, TemporalManager, SlotManager, Formatters):
+
+    def __init__(self,
+                 temporal_client: Client,
+                 couchbase_config: dict[str, Any],
+                 redis_config: dict[str, Any],
+                 logger: Logger,
+                 notification_handler: NotificationHandler):
+
+        # Initialize parent classes
+        Couchbase.__init__(self, connection_string=couchbase_config['connection_string'],
+                           username=couchbase_config['username'],
+                           password=couchbase_config['password'],
+                           logger=logger,
+                           notification_handler=notification_handler)
+
+        TemporalManager.__init__(self,
+                                 temporal_client=temporal_client,
+                                 logger=logger,
+                                 notification_handler=notification_handler)
+
+        SlotManager.__init__(self,
+                             host=redis_config['host'],
+                             port=redis_config['port'],
+                             username=redis_config['username'],
+                             password=redis_config['password'],
+                             logger=logger,
+                             notification_handler=notification_handler)
+
+        Formatters.__init__(self,
+                            logger=logger,
+                            notification_handler=notification_handler)
+
+    @activity.defn(name="prepare_activity")
+    async def prepare_activity(self, input_data: dict[str, Any]):
+        await super().prepare_activity(input_data)
+
+    def shutdown(self):
+        Couchbase.shutdown(self)

orchestrator/activities/couchbase.py

@@ -0,0 +1,94 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+    from logging import Logger
+    from datetime import timedelta
+    import json
+    import traceback
+    from couchbase.auth import PasswordAuthenticator
+    from couchbase.cluster import Cluster
+    from couchbase.options import ClusterOptions
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.temporal.activities.base import BaseActivity
+
+
+class Couchbase(BaseActivity):
+    def __init__(self, connection_string: str, username: str,
+                 password: str, logger: Logger,
+                 notification_handler: NotificationHandler):
+
+        self.connection_string = connection_string
+        self.username = username
+        self.password = password
+
+        logger.info("Initializing Couchbase connection...")
+        self.cluster = Cluster(
+            connection_string,
+            ClusterOptions(
+                authenticator=PasswordAuthenticator(
+                    username=username,
+                    password=password
+                )
+            )
+        )
+
+        logger.info("Awaiting Couchbase connection...")
+        self.cluster.wait_until_ready(timeout=timedelta(seconds=10))
+
+        logger.info("Couchbase connection ready")
+
+        BaseActivity.__init__(self,
+                              logger=logger,
+                              notification_handler=notification_handler)
+
+    def shutdown(self):
+        try:
+            self.cluster.close()
+        except Exception as e:
+            self.logger.error("Failed to close Couchbase connection: %s", e)
+
+    def __del__(self):
+        self.shutdown()
+
+    @activity.defn(name="load_query_from_couchbase")
+    async def load_query_from_couchbase(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Load a query from couchbase
+
+        Args:
+            input_data (dict[str, Any]): The input data containing the query to execute
+
+        Returns:
+            list[dict[str, Any]]: The result of the query
+        """
+        query = input_data['query']
+
+        self.logger.info("Executing couchbase query: %s", query)
+
+        try:
+            result = self.cluster.query(query)
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id="COUCHBASE_LOAD_QUERY_ERROR",
+                message=f"Failed to execute couchbase query: {e}",
+                block="load_query_from_couchbase",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+
+            self.logger.error(trace)
+            raise e
+
+        rows = []
+
+        for row in result.rows():
+            rows.append(row)
+
+        self.logger.info("Fetched %d rows from couchbase", len(rows))
+        self.logger.debug("Rows: \n %s",
+                          json.dumps(rows, indent=4, sort_keys=True))
+
+        return rows

orchestrator/activities/formatters.py

@@ -0,0 +1,362 @@
+from sientia_do.notifications.models import NotificationLevel
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import json
+    from typing import Any
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.activities.base import BaseActivity
+    from orchestrator.utils.orchestrator_functions import (
+        scouter, predictions_batch, gather_read_tags, build_tag_config
+    )
+    from math import ceil
+
+
+class Formatters(BaseActivity):
+    def __init__(self, logger: Logger, notification_handler: NotificationHandler):
+        BaseActivity.__init__(self, logger=logger,
+                              notification_handler=notification_handler)
+
+    @activity.defn(name="process_schedules")
+    async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Process schedules. Generates a schedule config dictionary
+        based on the input data workflow type.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to process.
+                - pipelines (list[dict[str, Any]]): The schedules to process.
+
+        Returns:
+            - dict[str, Any]: The schedule config dictionary
+        """
+
+        self.logger.info("Processing schedules...")
+
+        pipelines = input_data['pipelines']
+
+        schedule_config = {}
+
+        for pipeline in pipelines:
+            if pipeline['workflow_type'] == 'scouter':
+                schedule_config[pipeline['schedule_name']] = scouter(pipeline)
+            elif pipeline['workflow_type'] == 'predictions_batch':
+                schedule_config[pipeline['schedule_name']
+                                ] = predictions_batch(pipeline)
+
+        self.logger.info("Processed schedules")
+
+        self.logger.debug(json.dumps(
+            schedule_config, indent=4, sort_keys=True))
+
+        return schedule_config
+
+    @activity.defn(name="process_slots")
+    async def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Extracts all read tags from input pipelines, divides them into slots and
+        returns a slot config dictionary. If no ingestor is available, only one slot
+        is created.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to process.
+                - pipelines (list[dict[str, Any]]): The schedules to process.
+                - opc_servers (list[str]): The OPC servers to create ingestor config.
+                - active_ingestors (list[str]): The active ingestors to divide into slots.
+
+        Returns:
+            - dict[str, Any]: The slot config dictionary
+        """
+
+        self.logger.info("Processing slots...")
+
+        pipelines = input_data['pipelines']
+        opc_servers_list = input_data['opc_servers']
+        active_ingestors = input_data['active_ingestors']
+
+        opc_servers = {}
+        for server in opc_servers_list:
+            opc_servers[server['id']] = {
+                **server,
+            }
+
+        tags = list(gather_read_tags(pipelines).values())
+
+        number_of_tags = len(tags)
+        number_of_slots = len(active_ingestors) if active_ingestors else 1
+        tags_per_slot = ceil(number_of_tags / number_of_slots)
+
+        slot_config = {}
+        last_index = 0
+
+        for i in range(1, number_of_slots):
+            slot_config[f"{i}"] = {}
+            for tag in tags[last_index:last_index + tags_per_slot]:
+                slot_config = build_tag_config(
+                    tag, slot_config.copy(), opc_servers, i)
+            last_index += tags_per_slot
+
+        slot_config[f"{number_of_slots}"] = {}
+        for tag in tags[last_index:]:
+            slot_config = build_tag_config(
+                tag, slot_config.copy(), opc_servers, number_of_slots)
+
+        self.logger.info("Processed slots")
+        self.logger.debug(json.dumps(
+            slot_config, indent=4, sort_keys=True))
+
+        return slot_config
+
+    @activity.defn(name="create_schedule_config")
+    async def create_schedule_config(self,
+                                     input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Creates a schedule config dictionary based on the input data.
+        Checks the existing schedule config and updates it with the new schedule config,
+        deleting unnecessary schedules, creating new schedules and updating existing schedules.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to process.
+                - current_schedule_config (dict[str, Any]): The current schedule
+                config in Temporal server.
+                - schedule_config (dict[str, Any]): The schedule config to process.
+
+        Returns:
+            - dict[str, Any]: The schedule config dictionary
+        """
+
+        self.logger.info("Creating schedule config...")
+
+        current_schedule_config = input_data['current_schedule_config']
+        schedule_config = input_data['schedule_config']
+
+        to_update = {}
+        to_create = {}
+        to_delete = []
+
+        for schedule_name, schedule in schedule_config.items():
+            if schedule_name in current_schedule_config:
+                self.logger.debug(f"{current_schedule_config[schedule_name]}")
+                old_config = current_schedule_config[schedule_name]['data']
+
+                self.logger.debug(f"Comparing {schedule_name}:")
+                self.logger.debug(json.dumps(
+                    old_config, indent=4, sort_keys=True))
+                self.logger.debug(json.dumps(
+                    schedule, indent=4, sort_keys=True))
+
+                if schedule != old_config:
+                    to_update[schedule_name] = schedule
+
+            elif schedule_name not in current_schedule_config:
+                to_create[schedule_name] = schedule
+
+        for schedule_name in current_schedule_config:
+            if schedule_name not in schedule_config:
+                to_delete.append(schedule_name)
+
+        output = {
+            "to_update": to_update,
+            "to_create": to_create,
+            "to_delete": to_delete
+        }
+
+        self.logger.info("Created schedule config")
+        self.logger.debug(json.dumps(
+            output, indent=4, sort_keys=True))
+
+        return output
+
+    @activity.defn(name="create_slot_config")
+    async def create_slot_config(self,
+                                 input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Creates a slot config dictionary based on the input data.
+        Checks the existing slot config and updates it with the new slot config,
+        deleting unnecessary slots.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the slots to process.
+                - current_slot_config (dict[str, Any]): The current slot
+                config in Temporal server.
+                - slot_config (dict[str, Any]): The slot config to process.
+
+        Returns:
+            - dict[str, Any]: The slot config dictionary
+        """
+
+        self.logger.info("Creating slot config...")
+
+        current_slot_config = input_data['current_slot_config']
+        slot_config = input_data['slot_config']
+        to_delete = []
+
+        number_of_current_slots = len(current_slot_config)
+        number_of_slots = len(slot_config)
+
+        if number_of_current_slots > number_of_slots:
+            to_delete = [str(i) for i in range(
+                number_of_slots + 1, number_of_current_slots + 1)]
+
+        output = {
+            "to_delete": to_delete,
+            "to_insert": slot_config
+        }
+
+        self.logger.info("Created slot config")
+        self.logger.debug(json.dumps(
+            output, indent=4, sort_keys=True))
+
+        return output
+
+    def send_success_report(self, message: str, notification_id: str) -> None:
+        self.notification_handler.build_and_send_notification(
+            notification_id,
+            message,
+            "report_orchestration",
+            NotificationLevel.INFO
+        )
+
+    def send_error_report(self, message: str, notification_id: str,
+                          attachment: dict[str, Any]) -> None:
+        self.notification_handler.build_and_send_notification(
+            notification_id,
+            message,
+            "report_orchestration",
+            NotificationLevel.ERROR,
+            attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
+        )
+
+    def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]:
+        success_keys = [key for key, value
+                        in input_data.items() if value['success']]
+
+        error_keys = [key for key, value
+                      in input_data.items() if not value['success']]
+
+        return success_keys, error_keys
+
+    @activity.defn(name="report_schedule_orchestration")
+    async def report_schedule_orchestration(self,
+                                            input_data: dict[str, Any]) -> None:
+        """
+        Reports the orchestration result to the notification handler.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the orchestration result.
+                - created_schedules (dict[str, Any]): The created schedules.
+                - updated_schedules (dict[str, Any]): The updated schedules.
+                - deleted_schedules (list[str]): The deleted schedules.
+        """
+
+        self.logger.info("Reporting orchestration...")
+
+        created_schedules = input_data['created_schedules']
+        updated_schedules = input_data['updated_schedules']
+        deleted_schedules = input_data['deleted_schedules']
+
+        # Send report for created schedules
+        if len(created_schedules) > 0:
+            success_keys, error_keys = self.parse_report(created_schedules)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    f"Created schedules: \n {', '.join(success_keys)}",
+                    "REPORT_ORCHESTRATION_CREATED_SCHEDULES"
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    f"Failed to create schedules: \n {', '.join(error_keys)}",
+                    "REPORT_ORCHESTRATION_CREATED_SCHEDULES",
+                    created_schedules
+                )
+
+        # Send report for updated schedules
+        if len(updated_schedules) > 0:
+            success_keys, error_keys = self.parse_report(updated_schedules)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    f"Updated schedules: \n {', '.join(success_keys)}",
+                    "REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    f"Failed to update schedules: \n {', '.join(error_keys)}",
+                    "REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
+                    updated_schedules
+                )
+
+        if len(deleted_schedules) > 0:
+            success_keys, error_keys = self.parse_report(deleted_schedules)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    f"Deleted schedules: \n {', '.join(success_keys)}",
+                    "REPORT_ORCHESTRATION_DELETED_SCHEDULES"
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    f"Failed to delete schedules: \n {', '.join(error_keys)}",
+                    "REPORT_ORCHESTRATION_DELETED_SCHEDULES",
+                    deleted_schedules
+                )
+
+    @activity.defn(name="report_slot_orchestration")
+    async def report_slot_orchestration(self,
+                                        input_data: dict[str, Any]) -> None:
+        """
+        Reports the orchestration result to the notification handler.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the orchestration result.
+                - inserted_slots (dict[str, Any]): The inserted slots.
+                - deleted_slots (list[str]): The deleted slots.
+        """
+
+        self.logger.info("Reporting orchestration...")
+
+        inserted_slots = input_data['inserted_slots']
+        deleted_slots = input_data['deleted_slots']
+
+        if len(inserted_slots) > 0:
+            success_keys, error_keys = self.parse_report(inserted_slots)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    f"Inserted slots: \n {', '.join(success_keys)}",
+                    "REPORT_ORCHESTRATION_INSERTED_SLOTS"
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    f"Failed to insert slots: \n {', '.join(error_keys)}",
+                    "REPORT_ORCHESTRATION_INSERTED_SLOTS",
+                    inserted_slots
+                )
+
+        if len(deleted_slots) > 0:
+            success_keys, error_keys = self.parse_report(deleted_slots)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    f"Deleted slots: \n {', '.join(success_keys)}",
+                    "REPORT_ORCHESTRATION_DELETED_SLOTS"
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    f"Failed to delete slots: \n {', '.join(error_keys)}",
+                    "REPORT_ORCHESTRATION_DELETED_SLOTS",
+                    deleted_slots
+                )

orchestrator/activities/slot_manager.py

@@ -0,0 +1,153 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+    import json
+    from logging import Logger
+    from sientia_do.temporal.activities.redis_base import Redis
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+class SlotManager(Redis):
+
+    def __init__(self, host: str, port: int,
+                 username: str, password: str,
+                 logger: Logger, notification_handler: NotificationHandler):
+
+        Redis.__init__(self, host, port, username,
+                       password, logger, notification_handler)
+
+    @activity.defn(name="load_opc_slots")
+    async def load_opc_slots(self) -> dict[str, Any]:
+        """
+        Load all OPC slots from Redis
+
+        Returns:
+            dict[str, Any]: A dictionary of OPC slots
+        """
+
+        self.logger.info("Loading OPC slots...")
+
+        opc_slots = {}
+
+        slot_keys = self.redis_client.keys("slot:opc_tags:*")
+
+        self.logger.debug("Slot keys: %s", slot_keys)
+
+        if slot_keys:
+            if isinstance(slot_keys[0], bytes):
+                decoded_keys = [key.decode('utf-8') for key in slot_keys]
+            else:
+                decoded_keys = slot_keys
+
+            for key in decoded_keys:
+                opc_slots[key] = self.get(key)
+
+        self.logger.info(f"Loaded {len(opc_slots)} OPC slots")
+
+        self.logger.debug("Loaded: \n %s",
+                          json.dumps(opc_slots, indent=4, sort_keys=True))
+
+        return opc_slots
+
+    @activity.defn(name="load_active_ingestors")
+    async def load_active_ingestors(self) -> list[str]:
+        """
+        Load all active ingestors from Redis
+
+        Returns:
+            list[str]: A list of active ingestors
+        """
+
+        self.logger.info("Loading active ingestors...")
+
+        active_ingestors = self.redis_client.keys("heartbeat:ingestor:*")
+
+        self.logger.info(f"Loaded {len(active_ingestors)} active ingestors")
+
+        self.logger.debug("Active ingestors: \n %s", active_ingestors)
+
+        return [ingestor.decode('utf-8') for ingestor in active_ingestors]
+
+    @activity.defn(name="update_slots")
+    async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Update OPC slots in Redis
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the slots to update.
+                - to_insert (dict[str, Any]): The slots to insert.
+
+        Returns:
+            - report (dict[str, Any]): A report of the updated slots.
+        """
+
+        to_insert = input_data['to_insert']
+        self.logger.info("Updating OPC slots...")
+
+        report = {}
+
+        for slot in to_insert:
+            try:
+                self.set(f"slot:opc_tags:{slot}",
+                         to_insert[slot], ttl=None)
+                report[slot] = {
+                    "success": True,
+                    "message": "Slot updated successfully"
+                }
+            except Exception as e:
+                self.logger.error("Failed to update slot %s: %s",
+                                  slot, str(e))
+                report[slot] = {
+                    "success": False,
+                    "message": str(e)
+                }
+
+        self.logger.info(f"Updated {len(to_insert)} OPC slots")
+
+        self.logger.debug("Report: \n %s",
+                          json.dumps(report, indent=4, sort_keys=True))
+
+        return report
+
+    @activity.defn(name="delete_slots")
+    async def delete_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Delete OPC slots from Redis
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the slots to delete.
+                - to_delete (list[str]): The slots to delete.
+
+        Returns:
+            - report (dict[str, Any]): A report of the deleted slots.
+        """
+
+        to_delete = input_data['to_delete']
+        self.logger.info("Deleting OPC slots...")
+
+        report = {}
+
+        for slot in to_delete:
+            try:
+                self.redis_client.delete(f"slot:opc_tags:{slot}")
+                report[slot] = {
+                    "success": True,
+                    "message": "Slot deleted successfully"
+                }
+            except Exception as e:
+                self.logger.error("Failed to delete slot %s: %s",
+                                  slot, str(e))
+                report[slot] = {
+                    "success": False,
+                    "message": str(e)
+                }
+
+        self.logger.info(f"Deleted {len(to_delete)} OPC slots")
+
+        self.logger.debug("Report: \n %s",
+                          json.dumps(report, indent=4, sort_keys=True))
+
+        return report

orchestrator/activities/temporal_manager.py

@@ -0,0 +1,275 @@
+from temporalio import activity, workflow
+from temporalio.client import (
+    Client, Schedule, ScheduleActionStartWorkflow, ScheduleIntervalSpec, ScheduleSpec, ScheduleUpdate, ScheduleUpdateInput)
+from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.activities.base import BaseActivity
+    from google.protobuf.json_format import MessageToDict
+    import base64
+    from datetime import timedelta
+    import json
+    from orchestrator.utils.converters import parse_frequency
+
+
+class TemporalManager(BaseActivity):
+    def __init__(self, temporal_client: Client, logger: Logger,
+                 notification_handler: NotificationHandler):
+
+        self.temporal_client = temporal_client
+
+        self.schedule_handles = {}
+
+        self.model_id_id_key = SearchAttributeKey.for_keyword("model_id")
+        self.model_name_id_key = SearchAttributeKey.for_keyword("model_name")
+        self.orchestrated_id_key = SearchAttributeKey.for_keyword(
+            "orchestrated")
+
+        BaseActivity.__init__(self,
+                              logger=logger,
+                              notification_handler=notification_handler)
+
+    @activity.defn(name="load_schedule")
+    async def load_schedule(self) -> dict[str, Any]:
+        """
+        Load all orchestrated schedules from Temporal. Filters by search attribute
+        "Orchestrated" set to "true" and returns a dictionary of schedule_id:
+        {frequency, data, handle}
+
+        Returns:
+            dict[str, Any]: A dictionary of orchestrated schedules
+        """
+
+        self.logger.info("Getting orchestrated schedules...")
+
+        orchestrated_schedules = {}
+
+        async for schedule in await self.temporal_client.list_schedules():
+            search_attrs = getattr(schedule, "search_attributes", {})
+            if search_attrs.get("orchestrated", ["false"]) == ["true"]:
+                schedule_id = schedule.id
+
+                handle = self.temporal_client.get_schedule_handle(schedule_id)
+
+                self.schedule_handles[schedule_id] = handle
+
+                desc = await handle.describe()
+
+                for arg in desc.schedule.action.args:
+                    data = MessageToDict(arg)['data']
+                    data = base64.b64decode(data).decode('utf-8')
+
+                    frequency = desc.schedule.spec.intervals[0].every.seconds
+
+                    orchestrated_schedules[schedule_id] = {
+                        'frequency': frequency,
+                        'data': json.loads(data),
+                    }
+
+        self.logger.info("Found %d orchestrated schedules",
+                         len(orchestrated_schedules))
+
+        self.logger.debug("Orchestrated schedules: %s",
+                          orchestrated_schedules)
+
+        return orchestrated_schedules
+
+    @activity.defn(name="create_schedules")
+    async def create_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Create schedules in Temporal
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to create.
+                - schedules (dict[str, Any]): The schedules to create.
+
+        Returns:
+            - dict[str, Any]: A report of the created schedules.
+        """
+
+        schedules = input_data['schedules']
+
+        report = {}
+
+        for schedule_name, schedule in schedules.items():
+            search_attributes = TypedSearchAttributes([
+                SearchAttributePair(
+                    key=self.model_id_id_key,
+                    value=schedule['model_id']
+                ),
+                SearchAttributePair(
+                    key=self.model_name_id_key,
+                    value=schedule['model_name']
+                ),
+                SearchAttributePair(
+                    key=self.orchestrated_id_key,
+                    value="true"
+                )
+            ])
+            workflow_type = schedule['workflow_type']
+
+            try:
+                self.logger.debug(f"Creating schedule {schedule_name}:")
+                self.logger.debug(json.dumps(
+                    schedule, indent=4, sort_keys=True))
+                await self.temporal_client.create_schedule(
+                    schedule_name,
+                    Schedule(
+                        action=ScheduleActionStartWorkflow(
+                            workflow_type,
+                            schedule,
+                            id=schedule_name,
+                            task_queue=f"{workflow_type}-queue",
+                            execution_timeout=timedelta(minutes=2)
+                        ),
+                        spec=ScheduleSpec(
+                            intervals=[
+                                ScheduleIntervalSpec(
+                                    every=timedelta(seconds=parse_frequency(
+                                        schedule.get('frequency', '1m')))
+                                )
+                            ]
+                        )
+                    ),
+                    search_attributes=search_attributes
+                )
+
+                report[schedule_name] = {
+                    "success": True,
+                    "message": "Schedule created successfully"
+                }
+            except Exception as e:
+                self.logger.error("Failed to create schedule %s: %s",
+                                  schedule_name, str(e))
+                report[schedule_name] = {
+                    "success": False,
+                    "message": str(e)
+                }
+
+        self.logger.info(f"Processed {len(schedules)} schedules")
+
+        self.logger.debug("\n %s",
+                          json.dumps(report, indent=4, sort_keys=True))
+
+        return report
+
+    @activity.defn(name="update_schedules")
+    async def update_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Update schedules in Temporal
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to update.
+                - schedules (dict[str, Any]): The schedules to update.
+
+        Returns:
+            - dict[str, Any]: A report of the updated schedules.
+        """
+
+        schedules = input_data['schedules']
+
+        report = {}
+
+        for schedule_name, schedule in schedules.items():
+            try:
+                handler = self.schedule_handles.get(schedule_name)
+
+                if not handler:
+                    raise ValueError(f"Schedule {schedule_name} not found")
+
+                async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate:
+                    schedule_action = input_data.description.schedule.action
+
+                    self.logger.debug("Updating schedule:")
+
+                    if hasattr(schedule_action, "args"):
+                        self.logger.debug("New schedule:")
+                        self.logger.debug(json.dumps(
+                            schedule, indent=4, sort_keys=True))
+
+                        schedule_action.args = [schedule]
+
+                    input_data.description.schedule.spec.intervals = [
+                        ScheduleIntervalSpec(
+                            every=timedelta(seconds=parse_frequency(
+                                schedule.get('frequency', '1m')))
+                        )
+                    ]
+
+                    return ScheduleUpdate(schedule=input_data.description.schedule)
+
+                await handler.update(update_schedule)
+
+                del update_schedule
+
+                report[schedule_name] = {
+                    "success": True,
+                    "message": "Schedule updated successfully"
+                }
+            except Exception as e:
+                self.logger.error("Failed to update schedule %s: %s",
+                                  schedule_name, str(e))
+                report[schedule_name] = {
+                    "success": False,
+                    "message": str(e)
+                }
+
+        self.logger.info(f"Processed {len(schedules)} schedules")
+
+        self.logger.debug("\n %s",
+                          json.dumps(report, indent=4, sort_keys=True))
+
+        return report
+
+    @activity.defn(name="delete_schedules")
+    async def delete_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Delete schedules in Temporal
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to delete.
+                - schedules (list[str]): The schedules to delete.
+
+        Returns:
+            - dict[str, Any]: A report of the deleted schedules.
+        """
+
+        schedules = input_data['schedules']
+
+        report = {}
+
+        for schedule_name in schedules:
+            try:
+                handler = self.schedule_handles.get(schedule_name)
+
+                if not handler:
+                    raise ValueError(f"Schedule {schedule_name} not found")
+
+                await handler.delete()
+
+                del self.schedule_handles[schedule_name]
+
+                report[schedule_name] = {
+                    "success": True,
+                    "message": "Schedule deleted successfully"
+                }
+            except Exception as e:
+                self.logger.error("Failed to delete schedule %s: %s",
+                                  schedule_name, str(e))
+                report[schedule_name] = {
+                    "success": False,
+                    "message": str(e)
+                }
+
+        self.logger.info(f"Processed {len(schedules)} schedules")
+
+        self.logger.debug("\n %s",
+                          json.dumps(report, indent=4, sort_keys=True))
+
+        return report

orchestrator/utils/connectors_config.py

@@ -0,0 +1,18 @@
+from os import getenv
+
+
+def build_redis_config():
+    return {
+        'host': getenv('REDIS_HOST', 'localhost'),
+        'port': int(getenv('REDIS_PORT', '6379')),
+        'username': getenv('REDIS_USERNAME', None),
+        'password': getenv('REDIS_PASSWORD', None)
+    }
+
+
+def build_couchbase_config():
+    return {
+        'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'),
+        'username': getenv('COUCHBASE_USERNAME', 'sientia'),
+        'password': getenv('COUCHBASE_PASSWORD', 'sientia')
+    }

orchestrator/utils/converters.py

@@ -0,0 +1,14 @@
+def parse_frequency(frequency: str) -> int:
+    """
+    Parse frequency string to seconds
+    """
+    if frequency.endswith("s"):
+        return int(frequency[:-1])
+    elif frequency.endswith("m"):
+        return int(frequency[:-1]) * 60
+    elif frequency.endswith("h"):
+        return int(frequency[:-1]) * 60 * 60
+    elif frequency.endswith("d"):
+        return int(frequency[:-1]) * 60 * 60 * 24
+    else:
+        raise ValueError("Invalid frequency")

orchestrator/utils/orchestrator_functions.py

@@ -0,0 +1,164 @@
+from typing import Any
+
+
+def common_config(config: dict[str, Any]):
+    return {
+        "workflow_type": config['workflow_type'],
+        "schedule_name": config['schedule_name'],
+        "frequency": config.get('frequency', '1m'),
+        "max_retry_policy": config.get('max_retry_policy', 1),
+
+        "model_id": config['model_id'],
+        "model_name": config['models']['name'],
+    }
+
+
+def scouter(config: dict[str, Any]):
+    filters = {}
+    for f in config.get('filters', []):
+        filters[f['filter_name']] = {
+            "policy": f['policy']
+        }
+
+    tags = {}
+    for tag in config['read_tags']:
+        tags[tag['tag_name']] = {
+            "aggr_func": tag.get('aggr_func', 'lts'),
+            "data_range": tag.get('data_range', [-100, 100])
+        }
+
+    return {
+        **common_config(config),
+
+        "topic": f"raw_{config['schedule_name']}",
+        "trigger_laborious": False,
+        "filters": filters,
+        "schema": "sientia_data",
+        "table_name": "laborious_data",
+        "retention_time": config.get('tag_retention_minutes', 60) * 60,
+        "model_tags": tags
+    }
+
+
+def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
+    for fil in config:
+        base_filter_config[fil['filter_name']] = {
+            "policy": fil['policy'],
+            "config": fil.get('config', {})
+        }
+
+    return base_filter_config
+
+
+def process_path_priority(path_priority: list[str]):
+    for priority in path_priority[:]:
+        if priority not in ["STOP", "CONTINUE", "REPEAT"]:
+            path_priority.remove(priority)
+
+    for priority in ["STOP", "CONTINUE", "REPEAT"]:
+        if priority not in path_priority:
+            path_priority.append(priority)
+
+    return path_priority[0:3]
+
+
+def predictions_batch(config: dict[str, Any]):
+    tags = {}
+    for tag in config['write_tags']:
+        if tag['server_name'] not in tags:
+            tags[tag['server_name']] = {}
+
+        tag_type = tag['type']
+
+        if tag_type == 'prediction' or tag_type == 'confidence':
+            tag_type_str = f"{tag_type}_tags"
+
+            if tag_type_str not in tags[tag['server_name']]:
+                tags[tag['server_name']][tag_type_str] = {}
+
+            tags[tag['server_name']][tag_type_str][tag['addr']] = {
+                "data_type": tag.get('data_type', 'float'),
+            }
+
+    path_priority = process_path_priority(config.get(
+        'path_priority', ["STOP", "CONTINUE", "REPEAT"]))
+
+    return {
+        **common_config(config),
+
+        "query": config['query'],
+        "schema": "sientia_data",
+        "table_name": "predictions",
+        "retention_time": config.get('model_retention_minutes', 60) * 60,
+        "opc_output_config": tags,
+        "input_filters": overlap_filter_config({
+            "EMPTY_DATA": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['input_filters']),
+        "mlflow_transform_filters": overlap_filter_config({
+            "API_ERROR": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['mlflow_transform_filters']),
+        "mlflow_predict_filters": overlap_filter_config({
+            "API_ERROR": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['mlflow_predict_filters']),
+        "path_priority": path_priority
+    }
+
+
+def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
+    """
+    Gathers all read tags from input pipelines.
+
+    Args:
+        - pipelines (list[dict[str, Any]]): The schedules to process.
+
+    Returns:
+        - dict[str, Any]: The read tags dictionary
+    """
+
+    tags = {}
+
+    # Get all read tags from pipelines
+    for pipeline in pipelines:
+        for tag in pipeline['read_tags']:
+            tag_string = f"{tag['server_id']}:{tag['tag_address']}"
+            if tag_string not in tags:
+                tags[tag_string] = {
+                    **tag,
+                    "topics": []
+                }
+
+            tags[tag_string]['topics'].append(
+                f"raw_{pipeline['schedule_name']}")
+
+    return tags
+
+
+def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any],
+                     opc_servers: dict[str, Any], i: int):
+    server_id = tag['server_id']
+    server_name = opc_servers[server_id]['server_name']
+    if server_name not in slot_config[f"{i}"]:
+        slot_config[f"{i}"][server_name] = {
+            "server_id": server_id,
+            "name": server_name,
+            "url": opc_servers[server_id]['url'],
+            "server_uri": opc_servers[server_id]['uri'],
+            "tags": {}
+        }
+        for name, spec in opc_servers[server_id].get('security_spec', {}).items():
+            slot_config[f"{i}"][server_name][name] = spec
+
+    slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = {
+        **tag,
+    }
+
+    return slot_config

orchestrator/worker/worker.py

@@ -0,0 +1,103 @@
+from temporalio import workflow, client
+from temporalio.worker import Worker
+
+with workflow.unsafe.imports_passed_through():
+    import os
+    import sys
+    import asyncio
+    from orchestrator.workflows.orchestrator import Orchestrator
+    from orchestrator.activities.activities import Activities
+    from orchestrator.utils.connectors_config import (
+        build_couchbase_config,
+        build_redis_config,
+    )
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.utils.logger import get_logger
+
+
+async def main():
+    host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
+    logger = get_logger(__name__)
+
+    logger.info('Starting Worker...')
+
+    logger.info('Starting Notification Handler...')
+
+    notification_handler = NotificationHandler(
+        servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
+        logger=logger,
+        project_name=os.getenv('PROJECT_NAME', 'orchestrator'),
+    )
+
+    logger.info('Starting Temporal Client...')
+
+    temporal_client = await client.Client.connect(
+        target_host=host,
+        namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')
+    )
+
+    logger.info('Starting Activities...')
+
+    activities = Activities(
+        temporal_client=temporal_client,
+        couchbase_config=build_couchbase_config(),
+        redis_config=build_redis_config(),
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    logger.info('Starting Workers...')
+
+    workers = [
+        Worker(
+            temporal_client,
+            task_queue='orchestrator-queue',
+            workflows=[Orchestrator],
+            activities=[
+                # Base
+                activities.prepare_activity,
+                # Redis
+                activities.load_active_ingestors,
+                activities.load_opc_slots,
+                activities.update_slots,
+                activities.delete_slots,
+                # Couchbase
+                activities.load_query_from_couchbase,
+                # Temporal
+                activities.load_schedule,
+                activities.create_schedules,
+                activities.update_schedules,
+                activities.delete_schedules,
+                # Formatters
+                activities.process_schedules,
+                activities.process_slots,
+                activities.create_schedule_config,
+                activities.create_slot_config,
+                activities.report_schedule_orchestration,
+                activities.report_slot_orchestration,
+            ]
+        )
+    ]
+
+    handlers = []
+    for w in workers:
+        handlers.append(w.run())
+
+    logger.info('Workers started successfully')
+
+    try:
+        # This will run the workers and wait for them to complete.
+        # If an exception occurs in any of the worker handlers, it will be propagated here.
+        await asyncio.gather(*handlers)
+    except BaseException as e:
+        logger.error("An unhandled exception occurred: %s", e, exc_info=True)
+    finally:
+        if notification_handler:
+            notification_handler.shutdown()
+        if activities:
+            activities.shutdown()
+        # Exit with a non-zero status code to indicate failure to Kubernetes
+        sys.exit(1)
+
+if __name__ == '__main__':
+    asyncio.run(main())

orchestrator/workflows/orchestrator.py

@@ -0,0 +1,190 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from orchestrator.activities.activities import Activities
+    from typing import Any
+    from datetime import timedelta
+    from sientia_do.temporal.utils.policies import retry_policy
+
+
+@workflow.defn(name="orchestrator")
+class Orchestrator:
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+
+        input_data['workflow_name'] = 'orchestrator'
+
+        await workflow.execute_local_activity_method(
+            Activities.prepare_activity,
+            {
+                'workflow_name': input_data['workflow_name'],
+                'schedule_name': input_data['schedule_name'],
+                'model_name': '-',
+                'model_id': '-'
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        pipeline_config_handler = workflow.execute_local_activity_method(
+            Activities.load_query_from_couchbase,
+            {
+                'query': input_data['pipelines_query']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        opc_servers_handler = workflow.execute_local_activity_method(
+            Activities.load_query_from_couchbase,
+            {
+                'query': input_data['opc_servers_query']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        orchestrated_schedules_handler = workflow.execute_local_activity_method(
+            Activities.load_schedule,
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        current_slot_config_handler = workflow.execute_local_activity_method(
+            Activities.load_opc_slots,
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        active_ingestors_handler = workflow.execute_local_activity_method(
+            Activities.load_active_ingestors,
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        pipeline_config = await pipeline_config_handler
+        orchestrated_schedules = await orchestrated_schedules_handler
+        current_slot_config = await current_slot_config_handler
+        opc_servers = await opc_servers_handler
+        active_ingestors = await active_ingestors_handler
+
+        schedules_config_handler = workflow.execute_local_activity_method(
+            Activities.process_schedules,
+            {
+                'pipelines': pipeline_config
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        slot_config_handler = workflow.execute_local_activity_method(
+            Activities.process_slots,
+            {
+                'opc_servers': opc_servers,
+                'active_ingestors': active_ingestors,
+                'pipelines': pipeline_config,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        schedules_config = await schedules_config_handler
+        slot_config = await slot_config_handler
+
+        schedule_actions_handler = workflow.execute_local_activity_method(
+            Activities.create_schedule_config,
+            {
+                'current_schedule_config': orchestrated_schedules,
+                'schedule_config': schedules_config
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        slot_actions_handler = workflow.execute_local_activity_method(
+            Activities.create_slot_config,
+            {
+                'current_slot_config': current_slot_config,
+                'slot_config': slot_config
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        schedule_actions = await schedule_actions_handler
+        slot_actions = await slot_actions_handler
+
+        slot_deletion_report_handler = workflow.execute_activity_method(
+            Activities.delete_slots,
+            {
+                'to_delete': slot_actions['to_delete']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        slot_insertion_report_handler = workflow.execute_activity_method(
+            Activities.update_slots,
+            {
+                'to_insert': slot_actions['to_insert']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        schedule_deletion_report_handler = workflow.execute_activity_method(
+            Activities.delete_schedules,
+            {
+                'schedules': schedule_actions['to_delete']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        schedule_insertion_report_handler = workflow.execute_activity_method(
+            Activities.create_schedules,
+            {
+                'schedules': schedule_actions['to_create']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        schedule_update_report_handler = workflow.execute_activity_method(
+            Activities.update_schedules,
+            {
+                'schedules': schedule_actions['to_update']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        slot_deletion_report = await slot_deletion_report_handler
+        slot_insertion_report = await slot_insertion_report_handler
+        schedule_deletion_report = await schedule_deletion_report_handler
+        schedule_insertion_report = await schedule_insertion_report_handler
+        schedule_update_report = await schedule_update_report_handler
+
+        schedule_report_handler = workflow.execute_activity_method(
+            Activities.report_schedule_orchestration,
+            {
+                'created_schedules': schedule_insertion_report,
+                'updated_schedules': schedule_update_report,
+                'deleted_schedules': schedule_deletion_report
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        slot_report_handler = workflow.execute_activity_method(
+            Activities.report_slot_orchestration,
+            {
+                'inserted_slots': slot_insertion_report,
+                'deleted_slots': slot_deletion_report
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        await schedule_report_handler
+        await slot_report_handler

requirements.txt

@@ -0,0 +1,6 @@
+temporalio
+psycopg2-binary
+sqlalchemy
+redis
+couchbase
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14

samples.json

@@ -0,0 +1,69 @@
+{
+    "models": {
+        "1": {
+            "name": "Demo Model-Demo2"
+        }
+    },
+    "pipelines": {
+        "1": {
+            "schedule_name": "scouter-opcua-orchestrated-pipeline",
+            "model_id": "1",
+            "workflow_type": "scouter",
+            "frequency": "5s",
+            "max_retry_policy": 1,
+            "read_tags": [
+              {
+                "tag_name": "Counter",
+                "server_id": "default_server",
+                "aggr_func": "avg",
+                "tag_address": "ns=2;i=2",
+                "frequency": "15000",
+                "data_range": [
+                  -100,
+                  100
+                ]
+              },
+              {
+                "tag_name": "Rollout",
+                "server_id": "default_server",
+                "aggr_func": "mdn",
+                "tag_address": "ns=2;i=3",
+                "frequency": "15000",
+                "data_range": [
+                  -100,
+                  100
+                ]
+              },
+              {
+                "tag_name": "Square",
+                "server_id": "default_server",
+                "aggr_func": "lts",
+                "tag_address": "ns=2;i=4",
+                "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
+          }
+    },
+    "opc-servers": {
+        "1": {
+            "server_name": "default_server",
+            "url": "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840",
+            "uri": "http://opcua-server.simulator"
+        }
+    }
+}

test.ipynb

@@ -0,0 +1,408 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "id": "a287fa45",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Initializing Couchbase connection...\n",
+      "Awaiting Couchbase connection...\n",
+      "Couchbase connection ready\n"
+     ]
+    }
+   ],
+   "source": [
+    "from orchestrator.activities.couchbase import Couchbase\n",
+    "from unittest.mock import MagicMock\n",
+    "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
+    "couchbase = Couchbase(\n",
+    "    connection_string=\"couchbase://localhost\",\n",
+    "    username=\"sientia\",\n",
+    "    password=\"sientia\",\n",
+    "    logger=logger,\n",
+    "    notification_handler=MagicMock()\n",
+    ")\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "id": "d9a0f9c4",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Executing couchbase query: %s \n",
+      "SELECT\n",
+      "    pipelines.*,\n",
+      "    models as model\n",
+      "FROM\n",
+      "    `pipelines`\n",
+      "JOIN\n",
+      "    `models` ON KEYS pipelines.model_id;\n",
+      "\n",
+      "Fetched %d rows from couchbase 1\n",
+      "Rows: \n",
+      " %s [\n",
+      "    {\n",
+      "        \"filters\": [\n",
+      "            {\n",
+      "                \"filter_name\": \"OUT_OF_BOUNDS_FILTER\",\n",
+      "                \"policy\": \"DISCARD\"\n",
+      "            },\n",
+      "            {\n",
+      "                \"filter_name\": \"NULL_VALUES_FILTER\",\n",
+      "                \"policy\": \"DISCARD\"\n",
+      "            }\n",
+      "        ],\n",
+      "        \"frequency\": \"5s\",\n",
+      "        \"max_retry_policy\": 1,\n",
+      "        \"model\": {\n",
+      "            \"name\": \"Demo Model-Demo2\"\n",
+      "        },\n",
+      "        \"model_id\": \"1\",\n",
+      "        \"name\": \"scouter-opcua-pipeline\",\n",
+      "        \"read_tags\": [\n",
+      "            {\n",
+      "                \"aggr_func\": \"avg\",\n",
+      "                \"data_range\": [\n",
+      "                    -100,\n",
+      "                    100\n",
+      "                ],\n",
+      "                \"tag_name\": \"Counter\"\n",
+      "            }\n",
+      "        ],\n",
+      "        \"tag_retention_minutes\": 60,\n",
+      "        \"workflow_type\": \"scouter\"\n",
+      "    }\n",
+      "]\n",
+      "[{'filters': [{'filter_name': 'OUT_OF_BOUNDS_FILTER', 'policy': 'DISCARD'}, {'filter_name': 'NULL_VALUES_FILTER', 'policy': 'DISCARD'}], 'frequency': '5s', 'max_retry_policy': 1, 'model': {'name': 'Demo Model-Demo2'}, 'model_id': '1', 'name': 'scouter-opcua-pipeline', 'read_tags': [{'aggr_func': 'avg', 'data_range': [-100, 100], 'tag_name': 'Counter'}], 'tag_retention_minutes': 60, 'workflow_type': 'scouter'}]\n"
+     ]
+    }
+   ],
+   "source": [
+    "query = \"\"\"\n",
+    "SELECT\n",
+    "    pipelines.*,\n",
+    "    models as model\n",
+    "FROM\n",
+    "    `pipelines`\n",
+    "JOIN\n",
+    "    `models` ON KEYS pipelines.model_id;\n",
+    "\"\"\"\n",
+    "if __name__ == \"__main__\":\n",
+    "\n",
+    "\n",
+    "    result = await couchbase.load_query_from_couchbase({\n",
+    "        \"query\": query\n",
+    "    })\n",
+    "\n",
+    "    print(result)\n",
+    "    "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "id": "7d01f160",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from temporalio import client\n",
+    "from orchestrator.activities.temporal_manager import TemporalManager\n",
+    "import os\n",
+    "from unittest.mock import MagicMock\n",
+    "\n",
+    "host = \"localhost:7233\"\n",
+    "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
+    "\n",
+    "temporal_client = await client.Client.connect(\n",
+    "    target_host=host,\n",
+    "    namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n",
+    ")\n",
+    "\n",
+    "manager = TemporalManager(temporal_client=temporal_client,\n",
+    "                          logger=logger,\n",
+    "                          notification_handler=MagicMock())\n",
+    "                          "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "id": "bb750ae6",
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "<temporalio.client.ScheduleHandle at 0x7dfad094e610>"
+      ]
+     },
+     "execution_count": 6,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import asyncio\n",
+    "from datetime import timedelta\n",
+    "from temporalio.client import (\n",
+    "    Client,\n",
+    "    Schedule,\n",
+    "    ScheduleActionStartWorkflow,\n",
+    "    ScheduleIntervalSpec,\n",
+    "    ScheduleSpec,\n",
+    ")\n",
+    "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n",
+    "\n",
+    "\n",
+    "customer_id_key = SearchAttributeKey.for_keyword(\"orchestrated\")\n",
+    "search_attributes = TypedSearchAttributes([\n",
+    "    SearchAttributePair(customer_id_key, \"true\")\n",
+    "])\n",
+    "await temporal_client.create_schedule(\n",
+    "    \"meu-schedule-id5\",\n",
+    "    Schedule(\n",
+    "        action=ScheduleActionStartWorkflow(\n",
+    "            'scouter-test2',\n",
+    "            {\n",
+    "                'args': {\n",
+    "                    'arg1': 'value1'\n",
+    "                }\n",
+    "            },\n",
+    "            id=\"workflow-id-unico\",\n",
+    "            task_queue=\"nome-da-task-queue\",\n",
+    "        ),\n",
+    "        spec=ScheduleSpec(\n",
+    "            intervals=[ScheduleIntervalSpec(every=timedelta(minutes=10))]\n",
+    "        )\n",
+    "    ),\n",
+    "    search_attributes=search_attributes,\n",
+    ")\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "id": "1bd82225",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Getting orchestrated schedules...\n"
+     ]
+    }
+   ],
+   "source": [
+    "schedules = await manager.load_schedule()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 24,
+   "id": "1948670e",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from google.protobuf.json_format import MessageToDict\n",
+    "import base64\n",
+    "import json\n",
+    "for arg in schedules.schedule.action.args:\n",
+    "    data = MessageToDict(arg)\n",
+    "    data = base64.b64decode(data['data']).decode('utf-8')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 12,
+   "id": "a4c777dd",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from google.protobuf.json_format import MessageToDict\n",
+    "import base64\n",
+    "import json\n",
+    "\n",
+    "schedules_config = {}\n",
+    "async for schedule in await temporal_client.list_schedules():\n",
+    "    id = schedule.id\n",
+    "\n",
+    "    handle = temporal_client.get_schedule_handle(id)\n",
+    "\n",
+    "    desc = await handle.describe()\n",
+    "\n",
+    "    for arg in desc.schedule.action.args:\n",
+    "        data = MessageToDict(arg)['data']\n",
+    "        data = base64.b64decode(data).decode('utf-8')\n",
+    "\n",
+    "        frequency = desc.schedule.spec.intervals[0].every.seconds\n",
+    "\n",
+    "        schedules_config[id] = {\n",
+    "            'frequency': frequency,\n",
+    "            'data': json.loads(data),\n",
+    "            'handle': handle\n",
+    "        }\n",
+    "        \n",
+    "    \n",
+    "    \n",
+    "    "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 13,
+   "id": "988d1718",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "schedules_config = {\n",
+    "    \"scouter-opcua-orchestrated-pipeline\": schedules_config['scouter-opcua-orchestrated-pipeline'],\n",
+    "}"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 14,
+   "id": "c12f5e75",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "scouter-opcua-orchestrated-pipeline {'frequency': 5, 'data': {'filters': {'NULL_VALUES_FILTER': {'policy': 'DISCARD'}, 'OUT_OF_BOUNDS_FILTER': {'policy': 'DISCARD'}}, 'frequency': '5s', 'max_retry_policy': 1, 'model_id': '1', 'model_name': 'Demo Model-Demo2', 'model_tags': {'Counter': {'aggr_func': 'avg', 'data_range': [-100, 100]}}, 'retention_time': 3600, 'schedule_name': 'scouter-opcua-orchestrated-pipeline', 'schema': 'sientia_data', 'table_name': 'laborious_data', 'topic': 'raw_scouter-opcua-orchestrated-pipeline', 'trigger_laborious': False, 'workflow_type': 'scouter'}, 'handle': <temporalio.client.ScheduleHandle object at 0x7dfab3b2ab50>}\n",
+      "scouter-opcua-orchestrated-pipeline\n",
+      "{'workflow': 'scouter', 'args': [metadata {\n",
+      "  key: \"encoding\"\n",
+      "  value: \"json/plain\"\n",
+      "}\n",
+      "data: \"{\\\"filters\\\":{\\\"NULL_VALUES_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"},\\\"OUT_OF_BOUNDS_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"}},\\\"frequency\\\":\\\"5s\\\",\\\"max_retry_policy\\\":1,\\\"model_id\\\":\\\"1\\\",\\\"model_name\\\":\\\"Demo Model-Demo2\\\",\\\"model_tags\\\":{\\\"Counter\\\":{\\\"aggr_func\\\":\\\"avg\\\",\\\"data_range\\\":[-100,100]}},\\\"retention_time\\\":3600,\\\"schedule_name\\\":\\\"scouter-opcua-orchestrated-pipeline\\\",\\\"schema\\\":\\\"sientia_data\\\",\\\"table_name\\\":\\\"laborious_data\\\",\\\"topic\\\":\\\"raw_scouter-opcua-orchestrated-pipeline\\\",\\\"trigger_laborious\\\":false,\\\"workflow_type\\\":\\\"scouter\\\"}\"\n",
+      "], 'id': 'scouter-opcua-orchestrated-pipeline', 'task_queue': 'scouter-queue', 'execution_timeout': datetime.timedelta(seconds=120), 'run_timeout': None, 'task_timeout': None, 'retry_policy': None, 'memo': None, 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'headers': None, 'untyped_search_attributes': {}, 'static_summary': None, 'static_details': None, 'priority': Priority(priority_key=None)}\n"
+     ]
+    }
+   ],
+   "source": [
+    "from temporalio.client import Client, ScheduleUpdateInput, ScheduleUpdate, ScheduleSpec\n",
+    "\n",
+    "\n",
+    "for id, schedule in schedules_config.items():\n",
+    "    print(id, schedule)\n",
+    "    \n",
+    "    if schedule['frequency'] != 600:\n",
+    "        print(id)\n",
+    "\n",
+    "        handle = schedule['handle']\n",
+    "        \n",
+    "        async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n",
+    "            schedule_action = input.description.schedule.action\n",
+    "\n",
+    "            print(schedule_action.__dict__)\n",
+    "            \n",
+    "            if hasattr(schedule_action, 'args'):\n",
+    "                schedule_action.args = [{}]\n",
+    "            \n",
+    "            # Atualiza o intervalo de execução\n",
+    "            input.description.schedule.spec.intervals = [\n",
+    "                ScheduleIntervalSpec(every=timedelta(minutes=10))\n",
+    "            ]\n",
+    "            \n",
+    "            return ScheduleUpdate(schedule=input.description.schedule)\n",
+    "\n",
+    "        await handle.update(update_schedule)\n",
+    "            \n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "id": "87eb10c8",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from redis import Redis\n",
+    "\n",
+    "redis = Redis(host='localhost', port=6379)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "id": "fe4ad9dc",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Chave: slot:opc_tags:2, Valor: {\n",
+      "\"slot2\": \"value\"\n",
+      "}\n",
+      "Chave: slot:opc_tags:1, Valor: {\n",
+      "\"slot1\": \"value\"\n",
+      "}\n"
+     ]
+    }
+   ],
+   "source": [
+    "matching_keys = redis.keys(\"slot:opc_tags:*\")\n",
+    "\n",
+    "if matching_keys:\n",
+    "    decoded_keys = [key.decode('utf-8') for key in matching_keys]\n",
+    "    values = redis.mget(decoded_keys)\n",
+    "\n",
+    "    items = {}\n",
+    "    for i, key in enumerate(decoded_keys):\n",
+    "        value = values[i]\n",
+    "        if value is not None:\n",
+    "            try:\n",
+    "                items[key] = value.decode('utf-8')\n",
+    "            except (UnicodeDecodeError, AttributeError):\n",
+    "                items[key] = value\n",
+    "        else:\n",
+    "            items[key] = None\n",
+    "\n",
+    "    for key, value in items.items():\n",
+    "        print(f\"Chave: {key}, Valor: {value}\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "87a9dc89",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "venv",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

tests/orchestrator/activities/test_activities.py

@@ -0,0 +1,126 @@
+from unittest.mock import patch, MagicMock, ANY
+from pytest import mark
+from orchestrator.activities.activities import Activities
+from orchestrator.activities.couchbase import Couchbase
+from orchestrator.activities.temporal_manager import TemporalManager
+from orchestrator.activities.slot_manager import SlotManager
+from orchestrator.activities.formatters import Formatters
+
+
+@patch('orchestrator.activities.couchbase.Couchbase.__init__')
+@patch('orchestrator.activities.temporal_manager.TemporalManager.__init__')
+@patch('orchestrator.activities.slot_manager.SlotManager.__init__')
+@patch('orchestrator.activities.formatters.Formatters.__init__')
+def test___init__(mock_formatters_init, mock_slot_manager_init,
+                  mock_temporal_manager_init,
+                  mock_couchbase_init):
+
+    couchbase_config = {
+        'connection_string': 'couchbase://localhost',
+        'username': 'admin',
+        'password': 'password'
+    }
+
+    redis_config = {
+        'host': 'localhost',
+        'port': 6379,
+        'username': 'admin',
+        'password': 'password'
+    }
+
+    temporal_client = MagicMock()
+    logger = MagicMock()
+    notification_handler = MagicMock()
+
+    activities = Activities(
+        temporal_client=temporal_client,
+        couchbase_config=couchbase_config,
+        redis_config=redis_config,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    assert isinstance(activities, Activities)
+    assert isinstance(activities, Couchbase)
+    assert isinstance(activities, TemporalManager)
+    assert isinstance(activities, SlotManager)
+    assert isinstance(activities, Formatters)
+
+    mock_slot_manager_init.assert_called_once_with(
+        ANY,
+        host="localhost",
+        port=6379,
+        username="admin",
+        password="password",
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_couchbase_init.assert_called_once_with(
+        ANY,
+        connection_string=couchbase_config['connection_string'],
+        username=couchbase_config['username'],
+        password=couchbase_config['password'],
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_temporal_manager_init.assert_called_once_with(
+        ANY,
+        temporal_client=temporal_client,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_formatters_init.assert_called_once_with(
+        ANY,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+
+@mark.asyncio
+@patch('orchestrator.activities.couchbase.Cluster')
+async def test_prepare_activity(_mock_cluster):
+    couchbase_config = {
+        'connection_string': 'couchbase://localhost',
+        'username': 'admin',
+        'password': 'password'
+    }
+
+    redis_config = {
+        'host': 'localhost',
+        'port': 6379,
+        'username': 'admin',
+        'password': 'password'
+    }
+
+    temporal_client = MagicMock()
+    logger = MagicMock()
+    notification_handler = MagicMock()
+
+    activities = Activities(
+        temporal_client=temporal_client,
+        couchbase_config=couchbase_config,
+        redis_config=redis_config,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    input_data = {
+        'workflow_name': 'test-workflow-name',
+        'schedule_name': 'test-schedule-name',
+        'model_name': 'test-model-name',
+        'model_id': 'test-model-id'
+    }
+
+    await activities.prepare_activity(input_data)
+
+    assert activities.notification_handler.base_notification.pipeline == input_data[
+        'workflow_name']
+    assert activities.notification_handler.base_notification.trigger == input_data[
+        'schedule_name']
+    assert activities.notification_handler.base_notification.model_name == input_data[
+        'model_name']
+    assert activities.notification_handler.base_notification.model_id == input_data[
+        'model_id']

tests/orchestrator/activities/test_couchbase.py

@@ -0,0 +1,56 @@
+from unittest.mock import MagicMock, patch, ANY
+from pytest import fixture, mark, raises
+from sientia_do.notifications.models import NotificationLevel
+from orchestrator.activities.couchbase import Couchbase
+
+
+@fixture
+@patch("orchestrator.activities.couchbase.Cluster")
+def couchbase(_cluster_mock):
+    return Couchbase(
+        connection_string="couchbase://localhost",
+        username="admin",
+        password="password",
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+    )
+
+
+@mark.asyncio
+async def test_load_query_from_couchbase_success(couchbase):
+    couchbase.cluster.query.return_value.rows.return_value = [
+        {"id": "1", "name": "test"},
+        {"id": "2", "name": "test2"},
+    ]
+    query = "SELECT * FROM bucket"
+
+    result = await couchbase.load_query_from_couchbase({
+        "query": query,
+    })
+
+    assert result == [
+        {"id": "1", "name": "test"},
+        {"id": "2", "name": "test2"},
+    ]
+    couchbase.cluster.query.assert_called_once_with(query)
+    couchbase.notification_handler.build_and_send_notification.assert_not_called()
+
+
+@mark.asyncio
+async def test_load_query_from_couchbase_failure(couchbase):
+    couchbase.cluster.query.side_effect = Exception("Test error")
+    query = "SELECT * FROM bucket"
+
+    with raises(Exception):
+        await couchbase.load_query_from_couchbase({
+            "query": query,
+        })
+
+    couchbase.cluster.query.assert_called_once_with(query)
+    couchbase.notification_handler.build_and_send_notification.assert_called_once_with(
+        notification_id="COUCHBASE_LOAD_QUERY_ERROR",
+        message="Failed to execute couchbase query: Test error",
+        block="load_query_from_couchbase",
+        level=NotificationLevel.ERROR,
+        attachment_content=ANY,
+    )

tests/orchestrator/activities/test_formatters.py

@@ -0,0 +1,500 @@
+from unittest.mock import MagicMock, patch, call, ANY
+import json
+from pytest import fixture, mark
+from sientia_do.notifications.models import NotificationLevel
+from orchestrator.activities.formatters import Formatters
+from orchestrator.utils.orchestrator_functions import build_tag_config
+
+
+@fixture
+def formatters():
+    return Formatters(
+        logger=MagicMock(),
+        notification_handler=MagicMock()
+    )
+
+
+@mark.asyncio
+@patch("orchestrator.activities.formatters.scouter",
+       return_value="test_scouter")
+@patch("orchestrator.activities.formatters.predictions_batch",
+       return_value="test_predictions_batch")
+async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters):
+    input_data = {
+        "pipelines": [
+            {
+                "schedule_name": "test_schedule_name",
+                "workflow_type": "scouter",
+                "model_name": "test_model_name",
+                "model_id": "test_model_id"
+            },
+            {
+                "schedule_name": "test_schedule_name2",
+                "workflow_type": "predictions_batch",
+                "model_name": "test_model_name",
+                "model_id": "test_model_id"
+            }
+        ]
+    }
+
+    result = await formatters.process_schedules(input_data)
+
+    assert result == {
+        "test_schedule_name": "test_scouter",
+        "test_schedule_name2": "test_predictions_batch"
+    }
+
+    mock_scouter.assert_called_once_with(input_data['pipelines'][0])
+    mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1])
+
+
+@mark.asyncio
+@patch("orchestrator.activities.formatters.gather_read_tags",
+       return_value={
+           "1:test_tag_address": {
+               "server_id": "1",
+               "server_name": "test_server_name",
+               "tag_address": "test_tag_address",
+               "topics": ["raw_test_schedule"]
+           },
+           "2:test_tag_address2": {
+               "server_id": "2",
+               "server_name": "test_server_name2",
+               "tag_address": "test_tag_address2",
+               "topics": ["raw_test_schedule2"]
+           },
+           "2:test_tag_address3": {
+               "server_id": "2",
+               "server_name": "test_server_name2",
+               "tag_address": "test_tag_address3",
+               "topics": ["raw_test_schedule2"]
+           }
+       })
+@patch("orchestrator.activities.formatters.build_tag_config", side_effect=build_tag_config)
+async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters):
+    input_data = {
+        "opc_servers": [
+            {
+                "id": "1",
+                "server_name": "test_server_name",
+                "url": "test_url",
+                "uri": "test_uri",
+                "security_spec": {
+                    "test_name": "test_spec"
+                }
+            },
+            {
+                "id": "2",
+                "server_name": "test_server_name2",
+                "url": "test_url2",
+                "uri": "test_uri2"
+            }
+        ],
+        "active_ingestors": [
+            "test_active_ingestor1",
+            "test_active_ingestor2"
+        ],
+        "pipelines": "test_gather_read_tags"
+    }
+
+    result = await formatters.process_slots(input_data)
+
+    mock_gather_read_tags.assert_called_once_with(input_data['pipelines'])
+    mock_build_tag_config.assert_has_calls([
+        call(
+            {
+                "server_id": "1",
+                "server_name": "test_server_name",
+                "tag_address": "test_tag_address",
+                "topics": ["raw_test_schedule"]
+            },
+            ANY,
+            {
+                "1": {
+                    "id": "1",
+                    "server_name": "test_server_name",
+                    "url": "test_url",
+                    "uri": "test_uri",
+                    "security_spec": {
+                        "test_name": "test_spec"
+                    }
+                },
+                "2": {
+                    "id": "2",
+                    "server_name": "test_server_name2",
+                    "url": "test_url2",
+                    "uri": "test_uri2"
+                }
+            },
+            1
+        )
+    ])
+    mock_build_tag_config.assert_has_calls([
+        call(
+            {
+                "server_id": "2",
+                "server_name": "test_server_name2",
+                "tag_address": "test_tag_address2",
+                "topics": ["raw_test_schedule2"]
+            },
+            ANY,
+            {
+                "1": {
+                    "id": "1",
+                    "server_name": "test_server_name",
+                    "url": "test_url",
+                    "uri": "test_uri",
+                    "security_spec": {
+                        "test_name": "test_spec"
+                    }
+                },
+                "2": {
+                    "id": "2",
+                    "server_name": "test_server_name2",
+                    "url": "test_url2",
+                    "uri": "test_uri2"
+                }
+            },
+            1
+        )
+    ])
+    mock_build_tag_config.assert_has_calls([
+        call(
+            {
+                "server_id": "2",
+                "server_name": "test_server_name2",
+                "tag_address": "test_tag_address3",
+                "topics": ["raw_test_schedule2"]
+            },
+            ANY,
+            {
+                "1": {
+                    "id": "1",
+                    "server_name": "test_server_name",
+                    "url": "test_url",
+                    "uri": "test_uri",
+                    "security_spec": {
+                        "test_name": "test_spec"
+                    }
+                },
+                "2": {
+                    "id": "2",
+                    "server_name": "test_server_name2",
+                    "url": "test_url2",
+                    "uri": "test_uri2"
+                }
+            },
+            2
+        )
+    ])
+
+    assert result == {
+        "1": {
+            "test_server_name": {
+                "server_id": "1",
+                "name": "test_server_name",
+                "url": "test_url",
+                "server_uri": "test_uri",
+                "test_name": "test_spec",
+                "tags": {
+                    "test_tag_address": {
+                        "server_id": "1",
+                        "server_name": "test_server_name",
+                        "tag_address": "test_tag_address",
+                        "topics": ["raw_test_schedule"]
+                    }
+                }
+            },
+            "test_server_name2": {
+                "server_id": "2",
+                "name": "test_server_name2",
+                "url": "test_url2",
+                "server_uri": "test_uri2",
+                "tags": {
+                    "test_tag_address2": {
+                        "server_id": "2",
+                        "server_name": "test_server_name2",
+                        "tag_address": "test_tag_address2",
+                        "topics": ["raw_test_schedule2"]
+                    }
+                }
+            }
+        },
+        "2": {
+            "test_server_name2": {
+                "server_id": "2",
+                "name": "test_server_name2",
+                "url": "test_url2",
+                "server_uri": "test_uri2",
+                "tags": {
+                    "test_tag_address3": {
+                        "server_id": "2",
+                        "server_name": "test_server_name2",
+                        "tag_address": "test_tag_address3",
+                        "topics": ["raw_test_schedule2"]
+                    }
+                }
+            }
+        }
+    }
+
+
+@mark.asyncio
+async def test_create_schedule_config(formatters):
+    input_data = {
+        "current_schedule_config": {
+            "test_schedule_name_to_delete": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            },
+            "test_schedule_name_to_update": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            }
+        },
+        "schedule_config": {
+            "test_schedule_name_to_create": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            },
+            "test_schedule_name_to_update": {
+                "frequency": 60,
+                "data": {"test": "test2"}
+            }
+        }
+    }
+
+    result = await formatters.create_schedule_config(input_data)
+
+    assert result == {
+        "to_create": {
+            "test_schedule_name_to_create": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            }
+        },
+        "to_update": {
+            "test_schedule_name_to_update": {
+                "frequency": 60,
+                "data": {"test": "test2"}
+            }
+        },
+        "to_delete": [
+            "test_schedule_name_to_delete"
+        ]
+    }
+
+
+@mark.asyncio
+async def test_create_slot_config(formatters):
+    input_data = {
+        "current_slot_config": {
+            "1": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            },
+            "2": {
+                "frequency": 60,
+                "data": {"test": "test"}
+            }
+        },
+        "slot_config": {
+            "1": {
+                "frequency": 60,
+                "data": {"test": "test2"}
+            }
+        }
+    }
+
+    result = await formatters.create_slot_config(input_data)
+
+    assert result == {
+        "to_delete": [
+            "2"
+        ],
+        "to_insert": {
+            "1": {
+                "frequency": 60,
+                "data": {"test": "test2"}
+            }
+        }
+    }
+
+
+def test_send_success_report(formatters):
+    formatters.send_success_report("test_message", "test_notification_id")
+    formatters.notification_handler.build_and_send_notification.assert_called_once_with(
+        "test_notification_id",
+        "test_message",
+        "report_orchestration",
+        NotificationLevel.INFO
+    )
+
+
+def test_send_error_report(formatters):
+    formatters.send_error_report(
+        "test_message", "test_notification_id", {"test": "test"})
+    formatters.notification_handler.build_and_send_notification.assert_called_once_with(
+        "test_notification_id",
+        "test_message",
+        "report_orchestration",
+        NotificationLevel.ERROR,
+        attachment_content=json.dumps(
+            {"test": "test"}, indent=4, sort_keys=True)
+    )
+
+
+def test_parse_report(formatters):
+    input_data = {
+        "test_schedule_name_to_create": {
+            "success": True
+        },
+        "test_schedule_name_to_create_error": {
+            "success": False,
+            "error": "test_error"
+        }
+    }
+
+    result = formatters.parse_report(input_data)
+
+    assert result == (
+        ["test_schedule_name_to_create"],
+        ["test_schedule_name_to_create_error"]
+    )
+
+
+@mark.asyncio
+async def test_report_schedule_orchestration(formatters):
+    formatters.parse_report = MagicMock(
+        side_effect=formatters.parse_report
+    )
+    formatters.send_success_report = MagicMock()
+    formatters.send_error_report = MagicMock()
+
+    input_data = {
+        "created_schedules": {
+            "test_schedule_name_to_create": {
+                "success": True
+            },
+            "test_schedule_name_to_create_error": {
+                "success": False,
+                "error": "test_error"
+            }
+        },
+        "updated_schedules": {
+            "test_schedule_name_to_update": {
+                "success": True
+            },
+            "test_schedule_name_to_update_error": {
+                "success": False,
+                "error": "test_error"
+            }
+        },
+        "deleted_schedules": {
+            "test_schedule_name_to_delete": {
+                "success": True
+            },
+            "test_schedule_name_to_delete_error": {
+                "success": False,
+                "error": "test_error"
+            }
+        }
+    }
+
+    await formatters.report_schedule_orchestration(input_data)
+
+    formatters.parse_report.assert_has_calls([
+        call(input_data['created_schedules']),
+        call(input_data['updated_schedules']),
+        call(input_data['deleted_schedules'])
+    ])
+    formatters.send_success_report.assert_has_calls([
+        call(
+            "Created schedules: \n test_schedule_name_to_create",
+            "REPORT_ORCHESTRATION_CREATED_SCHEDULES"
+        ),
+        call(
+            "Updated schedules: \n test_schedule_name_to_update",
+            "REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
+        ),
+        call(
+            "Deleted schedules: \n test_schedule_name_to_delete",
+            "REPORT_ORCHESTRATION_DELETED_SCHEDULES"
+        )
+    ])
+    formatters.send_error_report.assert_has_calls([
+        call(
+            "Failed to create schedules: \n test_schedule_name_to_create_error",
+            "REPORT_ORCHESTRATION_CREATED_SCHEDULES",
+            input_data['created_schedules']
+        ),
+        call(
+            "Failed to update schedules: \n test_schedule_name_to_update_error",
+            "REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
+            input_data['updated_schedules']
+        ),
+        call(
+            "Failed to delete schedules: \n test_schedule_name_to_delete_error",
+            "REPORT_ORCHESTRATION_DELETED_SCHEDULES",
+            input_data['deleted_schedules']
+        )
+    ])
+
+
+@mark.asyncio
+async def test_report_slot_orchestration(formatters):
+    formatters.parse_report = MagicMock(
+        side_effect=formatters.parse_report
+    )
+    formatters.send_success_report = MagicMock()
+    formatters.send_error_report = MagicMock()
+
+    input_data = {
+        "inserted_slots": {
+            "test_slot_name_to_create": {
+                "success": True
+            },
+            "test_slot_name_to_create_error": {
+                "success": False,
+                "error": "test_error"
+            }
+        },
+        "deleted_slots": {
+            "test_slot_name_to_delete": {
+                "success": True
+            },
+            "test_slot_name_to_delete_error": {
+                "success": False,
+                "error": "test_error"
+            }
+        }
+    }
+
+    await formatters.report_slot_orchestration(input_data)
+
+    formatters.parse_report.assert_has_calls([
+        call(input_data['inserted_slots']),
+        call(input_data['deleted_slots'])
+    ])
+    formatters.send_success_report.assert_has_calls([
+        call(
+            "Inserted slots: \n test_slot_name_to_create",
+            "REPORT_ORCHESTRATION_INSERTED_SLOTS"
+        ),
+        call(
+            "Deleted slots: \n test_slot_name_to_delete",
+            "REPORT_ORCHESTRATION_DELETED_SLOTS"
+        )
+    ])
+    formatters.send_error_report.assert_has_calls([
+        call(
+            "Failed to insert slots: \n test_slot_name_to_create_error",
+            "REPORT_ORCHESTRATION_INSERTED_SLOTS",
+            input_data['inserted_slots']
+        ),
+        call(
+            "Failed to delete slots: \n test_slot_name_to_delete_error",
+            "REPORT_ORCHESTRATION_DELETED_SLOTS",
+            input_data['deleted_slots']
+        )
+    ])

tests/orchestrator/activities/test_slot_manager.py

@@ -0,0 +1,147 @@
+from unittest.mock import MagicMock, patch, call
+from pytest import mark, fixture
+from orchestrator.activities.slot_manager import SlotManager
+
+
+@fixture
+@patch("orchestrator.activities.slot_manager.Redis.__init__")
+def slot_manager(_redis_mock):
+
+    slot_manager = SlotManager(
+        host="localhost",
+        port=6379,
+        username="admin",
+        password="password",
+        logger=MagicMock(),
+        notification_handler=MagicMock()
+    )
+
+    slot_manager.redis_client = MagicMock()
+    slot_manager.logger = MagicMock()
+    slot_manager.notification_handler = MagicMock()
+
+    return slot_manager
+
+
+@mark.asyncio
+async def test_load_opc_slots_no_slot_keys(slot_manager):
+    slot_manager.redis_client.keys.return_value = []
+    assert await slot_manager.load_opc_slots() == {}
+
+
+@mark.asyncio
+async def test_load_opc_slots(slot_manager):
+    slot_manager.redis_client.keys.return_value = [
+        b"slot:opc_tags:1", b"slot:opc_tags:2", b"slot:opc_tags:3"]
+
+    slot_manager.get = MagicMock(
+        side_effect=[
+            "value1",
+            "value2",
+            None
+        ]
+    )
+
+    response = await slot_manager.load_opc_slots()
+
+    assert response == {
+        "slot:opc_tags:1": "value1",
+        "slot:opc_tags:2": "value2",
+        "slot:opc_tags:3": None
+    }
+
+
+@mark.asyncio
+async def test_load_opc_slots_no_decode(slot_manager):
+    slot_manager.redis_client.keys.return_value = [
+        "slot:opc_tags:1", "slot:opc_tags:2", "slot:opc_tags:3"]
+
+    slot_manager.get = MagicMock(
+        side_effect=[
+            "value1",
+            "value2",
+            None
+        ]
+    )
+
+    response = await slot_manager.load_opc_slots()
+
+    assert response == {
+        "slot:opc_tags:1": "value1",
+        "slot:opc_tags:2": "value2",
+        "slot:opc_tags:3": None
+    }
+
+
+@mark.asyncio
+async def test_load_active_ingestors(slot_manager):
+    slot_manager.redis_client.keys.return_value = [
+        b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", b"heartbeat:ingestor:3"]
+
+    response = await slot_manager.load_active_ingestors()
+
+    assert response == ["heartbeat:ingestor:1",
+                        "heartbeat:ingestor:2", "heartbeat:ingestor:3"]
+
+
+@mark.asyncio
+async def test_update_slots(slot_manager):
+    slot_manager.set = MagicMock(
+        side_effect=[
+            None,
+            Exception("Test exception")
+        ]
+    )
+
+    response = await slot_manager.update_slots({
+        "to_insert": {
+            "1": "value1",
+            "2": "value2"
+        }
+    })
+
+    slot_manager.set.assert_has_calls([
+        call("slot:opc_tags:1", "value1", ttl=None),
+        call("slot:opc_tags:2", "value2", ttl=None)
+    ])
+
+    assert response == {
+        "1": {
+            "success": True,
+            "message": "Slot updated successfully"
+        },
+        "2": {
+            "success": False,
+            "message": "Test exception"
+        }
+    }
+
+
+@mark.asyncio
+async def test_delete_slots(slot_manager):
+    slot_manager.redis_client.delete = MagicMock(
+        side_effect=[
+            None,
+            Exception("Test exception")
+        ]
+    )
+
+    response = await slot_manager.delete_slots({
+        "to_delete": ["1", "2"]
+    })
+
+    slot_manager.redis_client.delete.assert_has_calls([
+        call("slot:opc_tags:1"),
+        call("slot:opc_tags:2")
+    ])
+
+    assert response == {
+        "1": {
+            "success": True,
+            "message": "Slot deleted successfully"
+        },
+        "2": {
+            "success": False,
+            "message": "Test exception"
+        }
+    }

tests/orchestrator/activities/test_temporal_manager.py

@@ -0,0 +1,282 @@
+from unittest.mock import MagicMock, patch, AsyncMock, call, ANY
+from datetime import timedelta
+import base64
+import json
+from pytest import fixture, mark
+from orchestrator.activities.temporal_manager import TemporalManager
+from orchestrator.utils.converters import parse_frequency
+
+
+@fixture
+def temporal_manager():
+    return TemporalManager(
+        temporal_client=MagicMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock()
+    )
+
+
+@mark.asyncio
+@patch("orchestrator.activities.temporal_manager.MessageToDict",
+       return_value={"data": base64.b64encode(json.dumps({"test": "test"}).encode('utf-8'))})
+async def test_load_schedule(_mock_message_to_dict, temporal_manager):
+    # Create async iterator mock
+    async def async_iter():
+        yield MagicMock(
+            id="test-schedule-id",
+            search_attributes={
+                "orchestrated": ["true"]
+            }
+        )
+        yield MagicMock(
+            id="test-schedule-id-2",
+            search_attributes={
+                "Attr": ["false"]
+            }
+        )
+        yield MagicMock(
+            id="test-schedule-id-3",
+            search_attributes={
+                "Attr": ["false"]
+            }
+        )
+
+    temporal_manager.temporal_client.list_schedules = AsyncMock(
+        return_value=async_iter())
+    temporal_manager.temporal_client.get_schedule_handle.return_value = MagicMock(
+        describe=AsyncMock(
+            return_value=MagicMock(
+                schedule=MagicMock(
+                    action=MagicMock(
+                        args=[
+                            MagicMock(
+                                data=base64.b64encode(json.dumps(
+                                    {"test": "test"}).encode('utf-8'))
+                            )
+                        ]
+                    )
+                )
+            )
+        )
+    )
+    temporal_manager.temporal_client.get_schedule_handle.return_value.describe \
+        .return_value.schedule.spec = MagicMock(
+            intervals=[
+                MagicMock(
+                    every=MagicMock(
+                        seconds=60
+                    )
+                )
+            ]
+        )
+
+    response = await temporal_manager.load_schedule()
+
+    temporal_manager.temporal_client.list_schedules.assert_called_once()
+    assert response == {
+        "test-schedule-id": {
+            "frequency": 60,
+            "data": {"test": "test"}
+        }
+    }
+
+
+@mark.asyncio
+@patch("orchestrator.activities.temporal_manager.parse_frequency",
+       side_effect=parse_frequency)
+@patch("orchestrator.activities.temporal_manager.Schedule")
+@patch("orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow")
+@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec")
+@patch("orchestrator.activities.temporal_manager.ScheduleSpec")
+@patch("orchestrator.activities.temporal_manager.TypedSearchAttributes")
+@patch("orchestrator.activities.temporal_manager.SearchAttributePair")
+async def test_create_schedule(
+        mock_search_attribute_pair,
+        mock_typed_search_attributes,
+        mock_schedule_spec,
+        mock_schedule_interval_spec,
+        mock_schedule_action_start_workflow,
+        mock_schedule,
+        mock_parse_frequency,
+        temporal_manager):
+
+    input_data = {
+        "schedules": {
+            "test-schedule": {
+                "model_id": 1,
+                "model_name": "test-model-name",
+                "workflow_type": "test-workflow",
+                "frequency": "1m",
+                "data": {"test": "test"}
+            },
+            "test-schedule-invalid-frequency": {
+                "model_id": 2,
+                "model_name": "test-model-name",
+                "workflow_type": "test-workflow",
+                "frequency": "10y",
+                "data": {"test": "test"}
+            }
+        }
+    }
+
+    temporal_manager.temporal_client.create_schedule = AsyncMock()
+
+    report = await temporal_manager.create_schedules(input_data)
+
+    temporal_manager.temporal_client.create_schedule.assert_called_once_with(
+        "test-schedule",
+        mock_schedule.return_value,
+        search_attributes=mock_typed_search_attributes.return_value
+    )
+
+    mock_schedule.assert_called_once_with(
+        action=mock_schedule_action_start_workflow.return_value,
+        spec=mock_schedule_spec.return_value
+    )
+
+    mock_schedule_action_start_workflow.assert_has_calls([
+        call(
+            "test-workflow",
+            input_data['schedules']['test-schedule'],
+            id="test-schedule",
+            task_queue="test-workflow-queue",
+            execution_timeout=ANY
+        ),
+        call(
+            "test-workflow",
+            input_data['schedules']['test-schedule-invalid-frequency'],
+            id="test-schedule-invalid-frequency",
+            task_queue="test-workflow-queue",
+            execution_timeout=ANY
+        )
+    ])
+
+    mock_schedule_spec.assert_called_once_with(
+        intervals=[
+            mock_schedule_interval_spec.return_value
+        ]
+    )
+
+    mock_schedule_interval_spec.assert_called_once_with(
+        every=timedelta(seconds=60)
+    )
+
+    mock_parse_frequency.assert_has_calls([
+        call("1m"),
+        call("10y")
+    ])
+
+    mock_typed_search_attributes.assert_has_calls([
+        call([
+            mock_search_attribute_pair.return_value,
+            mock_search_attribute_pair.return_value,
+            mock_search_attribute_pair.return_value
+        ]),
+        call([
+            mock_search_attribute_pair.return_value,
+            mock_search_attribute_pair.return_value,
+            mock_search_attribute_pair.return_value
+        ])
+    ])
+
+    mock_search_attribute_pair.assert_has_calls([
+        call(
+            key=temporal_manager.model_id_id_key,
+            value=1
+        ),
+        call(
+            key=temporal_manager.model_name_id_key,
+            value="test-model-name"
+        ),
+        call(
+            key=temporal_manager.orchestrated_id_key,
+            value="true"
+        )
+    ])
+
+    assert report == {
+        "test-schedule": {
+            "success": True,
+            "message": "Schedule created successfully"
+        },
+        "test-schedule-invalid-frequency": {
+            "success": False,
+            "message": "Invalid frequency"
+        }
+    }
+
+
+@mark.asyncio
+@patch("orchestrator.activities.temporal_manager.parse_frequency",
+       side_effect=parse_frequency)
+@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec")
+async def test_update_schedules(
+        _mock_schedule_interval_spec,
+        _mock_parse_frequency,
+        temporal_manager):
+    input_mock = MagicMock(
+        args=MagicMock()
+    )
+    temporal_manager.schedule_handles = {
+        "test-schedule": MagicMock(
+            update=AsyncMock(
+                update=AsyncMock(
+                    side_effect=lambda f: f(input_mock)
+                )
+            )
+        )
+    }
+    input_data = {
+        "schedules": {
+            "test-schedule": {
+                "frequency": "1m",
+                "data": {"test": "test"}
+            },
+            "test-schedule_no_handler": {
+                "frequency": "1m",
+                "data": {"test": "test"}
+            }
+        }
+    }
+
+    report = await temporal_manager.update_schedules(input_data)
+
+    temporal_manager.schedule_handles['test-schedule'].update.assert_called_once()
+
+    assert report == {
+        "test-schedule": {
+            "success": True,
+            "message": "Schedule updated successfully"
+        },
+        "test-schedule_no_handler": {
+            "success": False,
+            "message": "Schedule test-schedule_no_handler not found"
+        }
+    }
+
+
+@mark.asyncio
+async def test_delete_schedules(temporal_manager):
+    temporal_manager.schedule_handles = {
+        "test-schedule": MagicMock(
+            delete=AsyncMock()
+        )
+    }
+    input_data = {
+        "schedules": [
+            "test-schedule", "test-schedule_no_handler"
+        ]
+    }
+
+    report = await temporal_manager.delete_schedules(input_data)
+
+    assert report == {
+        "test-schedule": {
+            "success": True,
+            "message": "Schedule deleted successfully"
+        },
+        "test-schedule_no_handler": {
+            "success": False,
+            "message": "Schedule test-schedule_no_handler not found"
+        }
+    }

tests/orchestrator/utils/test_connectors_config.py

@@ -0,0 +1,51 @@
+from os import environ
+from orchestrator.utils.connectors_config import (build_redis_config,
+                                                  build_couchbase_config)
+
+
+def test_build_redis_config_with_env_vars():
+    environ['REDIS_HOST'] = 'localhost'
+    environ['REDIS_PORT'] = '6379'
+    environ['REDIS_USERNAME'] = 'sientia'
+    environ['REDIS_PASSWORD'] = 'sientia'
+    assert build_redis_config() == {
+        'host': 'localhost',
+        'port': 6379,
+        'username': 'sientia',
+        'password': 'sientia'
+    }
+
+
+def test_build_couchbase_config_with_env_vars():
+    environ['COUCHBASE_CONNECTION_STRING'] = 'couchbase://localhost'
+    environ['COUCHBASE_USERNAME'] = 'sientia'
+    environ['COUCHBASE_PASSWORD'] = 'sientia'
+    assert build_couchbase_config() == {
+        'connection_string': 'couchbase://localhost',
+        'username': 'sientia',
+        'password': 'sientia'
+    }
+
+
+def test_build_redis_config_with_defaults():
+    environ.pop('REDIS_HOST', None)
+    environ.pop('REDIS_PORT', None)
+    environ.pop('REDIS_USERNAME', None)
+    environ.pop('REDIS_PASSWORD', None)
+    assert build_redis_config() == {
+        'host': 'localhost',
+        'port': 6379,
+        'username': None,
+        'password': None
+    }
+
+
+def test_build_couchbase_config_with_defaults():
+    environ.pop('COUCHBASE_CONNECTION_STRING', None)
+    environ.pop('COUCHBASE_USERNAME', None)
+    environ.pop('COUCHBASE_PASSWORD', None)
+    assert build_couchbase_config() == {
+        'connection_string': 'couchbase://localhost',
+        'username': 'sientia',
+        'password': 'sientia'
+    }

tests/orchestrator/utils/test_converters.py

@@ -0,0 +1,15 @@
+from orchestrator.utils.converters import parse_frequency
+
+
+def test_parse_frequency():
+    assert parse_frequency("1s") == 1
+    assert parse_frequency("1m") == 60
+    assert parse_frequency("1h") == 60 * 60
+    assert parse_frequency("1d") == 60 * 60 * 24
+
+    try:
+        parse_frequency("1")
+    except ValueError as e:
+        assert str(e) == "Invalid frequency"
+    else:
+        assert False

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -0,0 +1,343 @@
+from unittest.mock import patch, call
+from orchestrator.utils.orchestrator_functions import (
+    common_config,
+    scouter,
+    predictions_batch,
+    overlap_filter_config,
+    process_path_priority,
+    gather_read_tags,
+    build_tag_config
+)
+
+
+def test_common_config():
+    config = {
+        "workflow_type": "scouter",
+        "schedule_name": "test_schedule",
+        "model_id": "test_model_id",
+        "models": {
+            "name": "test_model_name"
+        }
+    }
+    result = common_config(config)
+    expected = {
+        "workflow_type": "scouter",
+        "schedule_name": "test_schedule",
+        "frequency": "1m",
+        "max_retry_policy": 1,
+        "model_id": "test_model_id",
+        "model_name": "test_model_name"
+    }
+    assert result == expected
+
+
+def test_scouter():
+    config = {
+        "workflow_type": "scouter",
+        "schedule_name": "test_schedule",
+        "model_id": "test_model_id",
+        "models": {
+            "name": "test_model_name"
+        },
+        "filters": [
+            {
+                "filter_name": "test_filter_name",
+                "policy": "test_policy"
+            }
+        ],
+        "read_tags": [
+            {
+                "tag_name": "test_tag_name",
+                "aggr_func": "test_aggr_func",
+                "data_range": [1, 2]
+            }
+        ],
+        "tag_retention_minutes": 10
+    }
+    result = scouter(config)
+    expected = {
+        "workflow_type": "scouter",
+        "schedule_name": "test_schedule",
+        "frequency": "1m",
+        "max_retry_policy": 1,
+        "model_id": "test_model_id",
+        "model_name": "test_model_name",
+        "topic": "raw_test_schedule",
+        "trigger_laborious": False,
+        "filters": {
+            "test_filter_name": {
+                "policy": "test_policy"
+            }
+        },
+        "schema": "sientia_data",
+        "table_name": "laborious_data",
+        "retention_time": 10 * 60,
+        "model_tags": {
+            "test_tag_name": {
+                "aggr_func": "test_aggr_func",
+                "data_range": [1, 2]
+            }
+        }
+    }
+    assert result == expected
+
+
+def test_overlap_filter_config():
+    config = [
+        {
+            "filter_name": "test_filter_name",
+            "policy": "test_policy"
+        },
+        {
+            "filter_name": "test_filter_name2",
+            "policy": "test_policy2"
+        }
+    ]
+    result = overlap_filter_config({
+        "test_filter_name": {
+            "policy": "test_policy"
+        }
+    }, config)
+    expected = {
+        "test_filter_name": {
+            "policy": "test_policy",
+            "config": {}
+        },
+        "test_filter_name2": {
+            "policy": "test_policy2",
+            "config": {}
+        }
+    }
+    assert result == expected
+
+
+def test_process_path_priority():
+    config = ["OTHER", "STOP", "CONTINUE"]
+    result = process_path_priority(config)
+    expected = ["STOP", "CONTINUE", "REPEAT"]
+    assert result == expected
+
+
+@patch('orchestrator.utils.orchestrator_functions.overlap_filter_config',
+       return_value={
+           "test_filter_name": {
+               "policy": "test_policy",
+               "config": {}
+           }
+       })
+@patch('orchestrator.utils.orchestrator_functions.process_path_priority',
+       return_value=["STOP", "CONTINUE", "REPEAT"])
+def test_predictions_batch(mock_process_path_priority,
+                           mock_overlap_filter_config):
+    config = {
+        "schedule_name": "test_schedule",
+        "workflow_type": "predictions_batch",
+        "model_id": "test_model_id",
+        "models": {
+            "name": "test_model_name"
+        },
+        "query": "test_query",
+        "write_tags": [
+            {
+                "server_name": "test_server_name",
+                "type": "prediction",
+                "addr": "test_addr"
+            },
+            {
+                "server_name": "test_server_name",
+                "type": "confidence",
+                "addr": "test_addr"
+            }
+        ],
+        "input_filters": [
+            {
+                "filter_name": "test_filter_name",
+                "policy": "test_policy"
+            }
+        ],
+        "mlflow_transform_filters": [
+            {
+                "filter_name": "test_filter_name",
+                "policy": "test_policy"
+            }
+        ],
+        "mlflow_predict_filters": [
+            {
+                "filter_name": "test_filter_name",
+                "policy": "test_policy"
+            }
+        ],
+        "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+    }
+
+    result = predictions_batch(config)
+
+    mock_overlap_filter_config.assert_has_calls([
+        call({
+            "EMPTY_DATA": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['input_filters'])
+    ])
+    mock_overlap_filter_config.assert_has_calls([
+        call({
+            "API_ERROR": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['mlflow_transform_filters'])
+    ])
+    mock_overlap_filter_config.assert_has_calls([
+        call({
+            "API_ERROR": {
+                "policy": "STOP",
+                "config": {}
+            }
+        }, config['mlflow_predict_filters'])
+    ])
+    mock_process_path_priority.assert_called_once_with(config['path_priority'])
+
+    expected = {
+        "workflow_type": "predictions_batch",
+        "schedule_name": "test_schedule",
+        "frequency": "1m",
+        "max_retry_policy": 1,
+        "model_id": "test_model_id",
+        "model_name": "test_model_name",
+        "query": "test_query",
+        "schema": "sientia_data",
+        "table_name": "predictions",
+        "retention_time": 60 * 60,
+        "opc_output_config": {
+            "test_server_name": {
+                "prediction_tags": {
+                    "test_addr": {
+                        "data_type": "float"
+                    }
+                },
+                "confidence_tags": {
+                    "test_addr": {
+                        "data_type": "float"
+                    }
+                }
+            }
+        },
+        "input_filters": {
+            "test_filter_name": {
+                "policy": "test_policy",
+                "config": {}
+            }
+        },
+        "mlflow_transform_filters": {
+            "test_filter_name": {
+                "policy": "test_policy",
+                "config": {}
+            }
+        },
+        "mlflow_predict_filters": {
+            "test_filter_name": {
+                "policy": "test_policy",
+                "config": {}
+            }
+        },
+        "path_priority": ["STOP", "CONTINUE", "REPEAT"]
+    }
+    assert result == expected
+
+
+def test_gather_read_tags():
+    pipelines = [
+        {
+            "schedule_name": "test_schedule",
+            "read_tags": [
+                {
+                    "server_id": "1",
+                    "server_name": "test_server_name",
+                    "tag_address": "test_tag_address"
+                }
+            ]
+        },
+        {
+            "schedule_name": "test_schedule2",
+            "read_tags": [
+                {
+                    "server_id": "2",
+                    "server_name": "test_server_name2",
+                    "tag_address": "test_tag_address2"
+                },
+                {
+                    "server_id": "2",
+                    "server_name": "test_server_name2",
+                    "tag_address": "test_tag_address3"
+                }
+            ]
+        }
+    ]
+
+    result = gather_read_tags(pipelines)
+
+    expected = {
+        "1:test_tag_address": {
+            "server_id": "1",
+            "server_name": "test_server_name",
+            "tag_address": "test_tag_address",
+            "topics": ["raw_test_schedule"]
+        },
+        "2:test_tag_address2": {
+            "server_id": "2",
+            "server_name": "test_server_name2",
+            "tag_address": "test_tag_address2",
+            "topics": ["raw_test_schedule2"]
+        },
+        "2:test_tag_address3": {
+            "server_id": "2",
+            "server_name": "test_server_name2",
+            "tag_address": "test_tag_address3",
+            "topics": ["raw_test_schedule2"]
+        }
+    }
+
+    assert result == expected
+
+
+def test_build_tag_config():
+    tag = {
+        "server_id": "1",
+        "server_name": "test_server_name",
+        "tag_address": "test_tag_address"
+    }
+    opc_servers = {
+        "1": {
+            "server_name": "test_server_name",
+            "url": "test_url",
+            "uri": "test_uri",
+            "security_spec": {
+                "test_name": "test_spec"
+            }
+        }
+    }
+    slot_config = {
+        "1": {}
+    }
+    i = 1
+    result = build_tag_config(tag, slot_config, opc_servers, i)
+    expected = {
+        "1": {
+            "test_server_name": {
+                "server_id": "1",
+                "name": "test_server_name",
+                "url": "test_url",
+                "server_uri": "test_uri",
+                "tags": {
+                    "test_tag_address": {
+                        "server_id": "1",
+                        "server_name": "test_server_name",
+                        "tag_address": "test_tag_address"
+                    }
+                },
+                "test_name": "test_spec"
+            }
+        }
+    }
+    assert result == expected

tests/orchestrator/workflows/test_orchestrator.py

@@ -0,0 +1,189 @@
+from unittest.mock import AsyncMock, patch, ANY, call
+from pytest import fixture, mark
+from orchestrator.workflows.orchestrator import Orchestrator
+from orchestrator.activities.activities import Activities
+
+
+@fixture
+def orchestrator():
+    return Orchestrator()
+
+
+@mark.asyncio
+@patch("orchestrator.workflows.orchestrator.workflow", new_callable=AsyncMock)
+async def test_run(workflow_mock, orchestrator):
+    input_data = {
+        "pipelines_query": "SELECT * FROM bucket",
+        "opc_servers_query": "SELECT * FROM servers",
+        "schedule_name": "test-schedule-name",
+    }
+
+    await orchestrator.run(input_data)
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.prepare_activity,
+            {
+                "workflow_name": "orchestrator",
+                "schedule_name": "test-schedule-name",
+                "model_name": "-",
+                "model_id": "-"
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.load_query_from_couchbase,
+            {
+                "query": input_data["pipelines_query"]
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.load_query_from_couchbase,
+            {
+                "query": input_data["opc_servers_query"]
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.load_schedule,
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.load_opc_slots,
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.load_active_ingestors,
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.process_schedules,
+            {
+                'pipelines': workflow_mock.execute_local_activity_method.return_value
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.process_slots,
+            {
+                'opc_servers': workflow_mock.execute_local_activity_method.return_value,
+                'active_ingestors': workflow_mock.execute_local_activity_method.return_value,
+                'pipelines': workflow_mock.execute_local_activity_method.return_value,
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.create_schedule_config,
+            {
+                'current_schedule_config': workflow_mock.execute_local_activity_method.return_value,
+                'schedule_config': workflow_mock.execute_local_activity_method.return_value
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.create_slot_config,
+            {
+                'current_slot_config': workflow_mock.execute_local_activity_method.return_value,
+                'slot_config': workflow_mock.execute_local_activity_method.return_value
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.delete_slots,
+            {
+                'to_delete':
+                    workflow_mock.execute_local_activity_method.return_value['to_delete']
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.update_slots,
+            {
+                'to_insert':
+                    workflow_mock.execute_local_activity_method.return_value['to_insert']
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.delete_schedules,
+            {
+                'schedules':
+                    workflow_mock.execute_local_activity_method.return_value['to_delete']
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.create_schedules,
+            {
+                'schedules':
+                    workflow_mock.execute_local_activity_method.return_value['to_create']
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.update_schedules,
+            {
+                'schedules':
+                    workflow_mock.execute_local_activity_method.return_value['to_update']
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])

values.yaml

@@ -0,0 +1,179 @@
+# Default values for sientia-module.
+# This is a YAML-formatted file.
+# 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: 1
+
+# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
+image:
+  repository: aignosi.azurecr.io/sientia-module
+  # This sets the pull policy for images.
+  pullPolicy: Always
+  # Overrides the image tag whose default is the chart appVersion.
+  tag: "0.1.0"
+
+# 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:
+- name: docker-hub-secret
+# This is to override the chart name.
+nameOverride: "sientia-orchestrator-worker"
+fullnameOverride: "sientia-orchestrator-worker"
+namespace: sientia
+
+# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
+serviceAccount:
+  # Specifies whether a service account should be created
+  create: true
+  # Automatically mount a ServiceAccount's API credentials?
+  automount: true
+  # Annotations to add to the service account
+  annotations: {}
+  # The name of the service account to use.
+  # If not set and create is true, a name is generated using the fullname template
+  name: "sientia-orchestrator-worker"
+
+# This is for setting Kubernetes Annotations to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
+podAnnotations: {}
+# This is for setting Kubernetes Labels to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+podLabels: {}
+
+podSecurityContext: {}
+  # fsGroup: 2000
+
+securityContext: {}
+  # capabilities:
+  #   drop:
+  #   - ALL
+  # readOnlyRootFilesystem: true
+  # runAsNonRoot: true
+  # runAsUser: 1000
+
+
+resources: {}
+  # We usually recommend not to specify default resources and to leave this as a conscious
+  # choice for the user. This also increases chances charts run on environments with little
+  # resources, such as Minikube. If you do want to specify resources, uncomment the following
+  # lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+  # limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  # requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+
+# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
+livenessProbe:
+  exec:
+    command:
+      - sh
+      - -c
+      - pgrep -f "orchestrator.worker.worker"
+  initialDelaySeconds: 20
+  periodSeconds: 30
+
+readinessProbe:
+  exec:
+    command:
+      - sh
+      - -c
+      - pgrep -f "orchestrator.worker.worker"
+  initialDelaySeconds: 10
+  periodSeconds: 15
+
+
+# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
+autoscaling:
+  enabled: false
+  minReplicas: 1
+  maxReplicas: 100
+  targetCPUUtilizationPercentage: 80
+  # targetMemoryUtilizationPercentage: 80
+
+# Additional volumes on the output Deployment definition.
+volumes: []
+# - name: foo
+#   secret:
+#     secretName: mysecret
+#     optional: false
+
+# Additional volumeMounts on the output Deployment definition.
+volumeMounts: []
+# - name: foo
+#   mountPath: "/etc/foo"
+#   readOnly: true
+
+nodeSelector: {}
+
+tolerations: []
+
+affinity: {}
+
+service:
+  enabled: false
+  type: ClusterIP
+  port: 4840
+  targetPort: 4840
+  
+
+env:
+  # Entrypoint variables 
+  - name: GITHUB_REPO_URL
+    value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
+  - name: GITHUB_BRANCH
+    value: "SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
+  - name: PYTHON_APP
+    value: "orchestrator.worker.worker"
+
+  # Application variables
+  - name: REDIS_HOST
+    value: "redis-master.redis.svc.cluster.local"
+  - name: REDIS_PORT
+    value: "6379"
+  - name: REDIS_USERNAME
+    valueFrom:
+      secretKeyRef:
+        name: redis
+        key: redis-username
+  - name: REDIS_PASSWORD
+    valueFrom:
+      secretKeyRef:
+        name: redis
+        key: redis-password
+
+  - name: COUCHBASE_CONNECTION_STRING
+    value: "couchbase://sientia.couchbase.svc.cluster.local"
+  - name: COUCHBASE_USERNAME
+    value: "sientia"
+  - name: COUCHBASE_PASSWORD
+    value: "sientia"
+
+  - name: KAFKA_BOOTSTRAP_SERVERS
+    value: "kafka.kafka.svc.cluster.local:9092"
+
+  - name: LOG_LEVEL
+    value: "DEBUG"
+  - name: PROJECT_NAME
+    value: "sientia-orchestrator"
+
+  - name: TEMPORAL_HOST
+    value: "temporal-frontend.temporal.svc.cluster.local:7233"
+  - name: TEMPORAL_NAMESPACE
+    value: "default"
+
+ssh:
+  enabled: true
+  secretName: git-ssh-key-sientia-orchestrator-worker
+  sshPath: /mnt/.ssh
+  knownHostsPath: /mnt/known_hosts
+
+# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
+
+# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat
+
+# kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \
+#   --namespace sientia \
+#   --from-file=ssh-privatekey=git_key \
+#   --type=kubernetes.io/ssh-auth