{ "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": 3, "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", " " ] }, { "cell_type": "code", "execution_count": null, "id": "bb750ae6", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "" ] }, "execution_count": 4, "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", "# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n", "\n", "\n", "\n", "await temporal_client.create_schedule(\n", " \"orchestrator\",\n", " Schedule(\n", " action=ScheduleActionStartWorkflow(\n", " 'orchestrator',\n", " {\n", " \"schedule_name\": \"orchestrator-test\",\n", " \"pipelines_query\": {\n", " \"collection\": \"pipelines\",\n", " \"aggregation\": [\n", " {\n", " \"$lookup\": {\n", " \"from\": \"models\",\n", " \"localField\": \"model_id\",\n", " \"foreignField\": \"id\",\n", " \"as\": \"model_docs\"\n", " }\n", " },\n", " {\n", " \"$match\": {\n", " \"active\": True\n", " }\n", " },\n", " {\n", " \"$addFields\": {\n", " \"models\": {\n", " \"$arrayElemAt\": [\n", " \"$model_docs\",\n", " 0\n", " ]\n", " }\n", " }\n", " },\n", " {\n", " \"$match\": {\n", " \"models.active\": True\n", " }\n", " },\n", " {\n", " \"$project\": {\n", " \"model_docs\": 0\n", " }\n", " }\n", " ]\n", " },\n", " \"opc_servers_query\": {\n", " \"collection\": \"opc-servers\",\n", " \"filters\": {\n", "\n", " }\n", " }\n", " },\n", " id=\"orchestrator\",\n", " task_queue=\"orchestrator-queue\",\n", " execution_timeout=timedelta(minutes=600)\n", " ),\n", " spec=ScheduleSpec(\n", " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n", " )\n", " )\n", ")\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b840347d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Error deleting schedule scouter-load-test-1: workflow execution already completed\n", "Error deleting schedule scouter-load-test-2: workflow execution already completed\n", "Error deleting schedule scouter-load-test-3: workflow execution already completed\n", "Error deleting schedule scouter-load-test-4: workflow execution already completed\n", "Error deleting schedule scouter-load-test-5: workflow execution already completed\n", "Error deleting schedule scouter-load-test-6: workflow execution already completed\n", "Error deleting schedule scouter-load-test-7: workflow execution already completed\n", "Error deleting schedule scouter-load-test-8: workflow execution already completed\n", "Error deleting schedule scouter-load-test-9: workflow execution already completed\n", "Error deleting schedule scouter-load-test-10: workflow execution already completed\n", "Error deleting schedule scouter-load-test-11: workflow execution already completed\n", "Error deleting schedule scouter-load-test-12: workflow execution already completed\n", "Error deleting schedule scouter-load-test-13: workflow execution already completed\n", "Error deleting schedule scouter-load-test-14: workflow execution already completed\n", "Error deleting schedule scouter-load-test-15: workflow execution already completed\n", "Error deleting schedule scouter-load-test-16: workflow execution already completed\n", "Error deleting schedule scouter-load-test-17: workflow execution already completed\n", "Error deleting schedule scouter-load-test-18: workflow execution already completed\n", "Error deleting schedule scouter-load-test-19: workflow execution already completed\n" ] } ], "source": [ "# Delete schedule by ID\n", "schedule_id = \"scouter-load-test-num\"\n", "\n", "for i in range(0, 70):\n", "\n", " try:\n", " handle = temporal_client.get_schedule_handle(\n", " schedule_id.replace(\"num\", str(i)))\n", " \n", " await handle.delete()\n", " except Exception as e:\n", " print(f\"Error deleting schedule {schedule_id.replace('num', str(i))}: {e}\")\n", "\n", "# Delete schedule by ID\n", "schedule_id = \"laborious-load-test-num\"\n", "\n", "for i in range(0, 70):\n", "\n", " try:\n", " handle = temporal_client.get_schedule_handle(\n", " schedule_id.replace(\"num\", str(i)))\n", " \n", " await handle.delete()\n", " except Exception as e:\n", " print(f\"Error deleting schedule {schedule_id.replace('num', str(i))}: {e}\")\n" ] }, { "cell_type": "code", "execution_count": 7, "id": "374b5b0e", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "scouter-load-test-num\n" ] } ], "source": [ "schedule_id = \"scouter-load-test-num\"\n", "schedule_id.replace(\"-\", \"_\")\n", "\n", "print(schedule_id)" ] }, { "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": 4, "id": "f81b3728", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "{'id': 'orchestrator', 'schedule': ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='orchestrator'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=3600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), 'info': ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 9, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 9, 0, 0, 165083, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T09:00:00Z', first_execution_run_id='01980829-9700-771a-9c9d-2450eecc9af4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 10, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 10, 0, 0, 125309, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T10:00:00Z', first_execution_run_id='01980860-8578-759d-97fa-dbdfc784f88a')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 11, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 11, 0, 0, 124356, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T11:00:00Z', first_execution_run_id='01980897-73f7-7e83-a7a6-20eaac38e7d4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 12, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 12, 0, 0, 167329, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T12:00:00Z', first_execution_run_id='019808ce-62a2-75d6-b11d-c5387cfa258f')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 13, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 13, 0, 0, 167575, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T13:00:00Z', first_execution_run_id='01980905-5122-7807-90ae-3782d0332e79'))], next_action_times=[datetime.datetime(2025, 7, 14, 14, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 15, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 16, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 17, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 18, 0, tzinfo=datetime.timezone.utc)]), 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'search_attributes': {}, 'data_converter': DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), 'raw_entry': schedule_id: \"orchestrator\"\n", "info {\n", " spec {\n", " interval {\n", " interval {\n", " seconds: 3600\n", " }\n", " }\n", " }\n", " workflow_type {\n", " name: \"orchestrator\"\n", " }\n", " recent_actions {\n", " schedule_time {\n", " seconds: 1752483600\n", " }\n", " actual_time {\n", " seconds: 1752483600\n", " nanos: 165083498\n", " }\n", " start_workflow_result {\n", " workflow_id: \"orchestrator-2025-07-14T09:00:00Z\"\n", " run_id: \"01980829-9700-771a-9c9d-2450eecc9af4\"\n", " }\n", " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", " }\n", " recent_actions {\n", " schedule_time {\n", " seconds: 1752487200\n", " }\n", " actual_time {\n", " seconds: 1752487200\n", " nanos: 125309119\n", " }\n", " start_workflow_result {\n", " workflow_id: \"orchestrator-2025-07-14T10:00:00Z\"\n", " run_id: \"01980860-8578-759d-97fa-dbdfc784f88a\"\n", " }\n", " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", " }\n", " recent_actions {\n", " schedule_time {\n", " seconds: 1752490800\n", " }\n", " actual_time {\n", " seconds: 1752490800\n", " nanos: 124356906\n", " }\n", " start_workflow_result {\n", " workflow_id: \"orchestrator-2025-07-14T11:00:00Z\"\n", " run_id: \"01980897-73f7-7e83-a7a6-20eaac38e7d4\"\n", " }\n", " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", " }\n", " recent_actions {\n", " schedule_time {\n", " seconds: 1752494400\n", " }\n", " actual_time {\n", " seconds: 1752494400\n", " nanos: 167329351\n", " }\n", " start_workflow_result {\n", " workflow_id: \"orchestrator-2025-07-14T12:00:00Z\"\n", " run_id: \"019808ce-62a2-75d6-b11d-c5387cfa258f\"\n", " }\n", " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", " }\n", " recent_actions {\n", " schedule_time {\n", " seconds: 1752498000\n", " }\n", " actual_time {\n", " seconds: 1752498000\n", " nanos: 167575982\n", " }\n", " start_workflow_result {\n", " workflow_id: \"orchestrator-2025-07-14T13:00:00Z\"\n", " run_id: \"01980905-5122-7807-90ae-3782d0332e79\"\n", " }\n", " start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n", " }\n", " future_action_times {\n", " seconds: 1752501600\n", " }\n", " future_action_times {\n", " seconds: 1752505200\n", " }\n", " future_action_times {\n", " seconds: 1752508800\n", " }\n", " future_action_times {\n", " seconds: 1752512400\n", " }\n", " future_action_times {\n", " seconds: 1752516000\n", " }\n", "}\n", "}\n", "{'_client': , 'id': 'orchestrator'}\n" ] } ], "source": [ "async for schedule in await temporal_client.list_schedules():\n", " print(vars(schedule))\n", " id = schedule.id\n", "\n", " handle = temporal_client.get_schedule_handle(id)\n", "\n", " print(vars(handle))" ] }, { "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': }\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 }