Merge pull request #22 from imubit/service-mode-update

cloud-rocket · web-flow · commit 52bf157887cc · 2025-04-14T10:59:17.000-05:00
Service mode update
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -62,8 +62,8 @@ jobs:
     strategy:
       matrix:
         python:
-        - "3.8"  # oldest Python supported by PSF
-        - "3.11"  # newest Python that is stable
+        - "3.11"  # oldest Python supported by PSF
+        - "3.13"  # newest Python that is stable
         platform:
         - ubuntu-latest
 #        - windows-latest
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -2,7 +2,7 @@ exclude: '^docs/conf.py'
 
 repos:
 - repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v4.4.0
+  rev: v5.0.0
   hooks:
   - id: trailing-whitespace
   - id: check-added-large-files
diff --git a/README.md b/README.md
@@ -40,6 +40,7 @@ with LocalAgent() as agent:
 
 ## Command Line Usage
 
+Start broker based agent
 ```bash
-dagent --service.id my_service --broker.uri amqp://guest:guest@192.168.4.23/
+da-broker-svc --service.id my_service --broker.uri amqp://guest:guest@192.168.4.23/
 ```
diff --git a/docs/cli-usage.md b/docs/cli-usage.md
@@ -0,0 +1,15 @@
+# CLI Usage
+
+## Creating OPC UA Connection
+
+### Matrikon simulator
+
+```commandline
+da-broker exec create_connection --conn_type=opc-ua --conn_name=ua --server_uri=opc.tcp://127.0.0.1:4840/freeopcua/server/
+da-broker exec enable_connection --conn_name=ua
+```
+
+### Matrikon simulator
+```
+da-broker exec create_job --job_id=job1 --conn_name=ua --tags="['2:PLC Server/2:CyclicData', '2:PLC Server/2:BooleanData']" --seconds=2
+```
diff --git a/setup.cfg b/setup.cfg
@@ -70,8 +70,9 @@ testing =
 
 [options.entry_points]
 console_scripts =
-    dagentsvc = data_agent.broker_agent_main:run_event_loop
-    dagent = data_agent.cli_main:run
+    da-broker-svc = data_agent.broker.agent_main:run_event_loop
+    da-broker = data_agent.broker.cli:run
+    data-agent = data_agent.local_cli:run
 pyinstaller40 =
     hook-dirs = data_agent.win32.hooks:get_hook_dirs
 
@@ -97,6 +98,8 @@ testpaths = tests
 # markers =
 #     slow: mark tests as slow (deselect with '-m "not slow"')
 
+asyncio_default_fixture_loop_scope = function
+
 [aliases]
 dists = bdist_wheel
 
diff --git a/src/data_agent/abstract_connector.py b/src/data_agent/abstract_connector.py
@@ -129,25 +129,3 @@ def read_tag_values_period(
     @abstractmethod
     def write_tag_values(self, tags: dict, wait_for_result: bool, **kwargs) -> dict:
         pass
-
-    # @abstractmethod
-    # def list_groups(self) -> list:
-    #     pass
-    #
-    # @abstractmethod
-    # def register_group(self, group_name: str, tags: list, refresh_rate_ms: int = 1000):
-    #     pass
-    #
-    # @abstractmethod
-    # def unregister_group(self, group_name: str):
-    #     pass
-    #
-    # @abstractmethod
-    # def read_group_values(self, group_name: str, from_cache: bool = True) -> dict:
-    #     pass
-    #
-    # @abstractmethod
-    # def write_group_values(
-    #     self, group_name: str, tags: dict, wait_for_result: bool, **kwargs
-    # ) -> dict:
-    #     pass
diff --git a/src/data_agent/broker/__init__.py b/src/data_agent/broker/__init__.py
diff --git a/src/data_agent/broker/agent.py b/src/data_agent/broker/agent.py
@@ -20,6 +20,7 @@
 )
 from data_agent.connection_manager import ConnectionManager
 from data_agent.daq_scheduler import create_daq_scheduler
+from data_agent.exchanger import DataExchanger
 from data_agent.safe_manipulator import SafeManipulator
 
 log = logging.getLogger(__name__)
@@ -30,6 +31,7 @@ class BrokerAgent:
     _broker_conn = None
     _connection_manager = None
     _safe_manipulator = None
+    _data_exchanger = None
     _scheduler = None
 
     async def init(self, loop, is_service=False, enable_persistance=True):
@@ -92,8 +94,12 @@ async def init(self, loop, is_service=False, enable_persistance=True):
                 enable_persistence=enable_persistance,
             ),
         )
+        self._data_exchanger = DataExchanger(self._connection_manager)
         api = ServiceApi(
-            self._scheduler, self._connection_manager, self._safe_manipulator
+            self._scheduler,
+            self._connection_manager,
+            self._data_exchanger,
+            self._safe_manipulator,
         )
         await self._broker_conn.rpc_register(api)
 
@@ -102,7 +108,7 @@ async def init(self, loop, is_service=False, enable_persistance=True):
             "************ Data Agent Service Initialized *************************"
         )
         log.info(
-            f" Supported connectors: {self._connection_manager.list_supported_connectors()}"
+            f" Supported connectors: {list(self._connection_manager.list_supported_connectors().keys())}"
         )
         log.info(
             "***********************************************************************"
@@ -117,6 +123,7 @@ async def close(self):
         self._broker_conn = None
         self._connection_manager = None
         self._safe_manipulator = None
+        self._data_exchanger = None
         self._scheduler = None
         log.info("")
         log.info(
diff --git a/src/data_agent/broker/agent_main.py b/src/data_agent/broker/agent_main.py
@@ -1,7 +1,7 @@
 import asyncio
 import logging
 
-from data_agent.broker_agent import BrokerAgent
+from data_agent.broker.agent import BrokerAgent
 
 log = logging.getLogger(__name__)
 
diff --git a/src/data_agent/broker/cli.py b/src/data_agent/broker/cli.py
@@ -0,0 +1,78 @@
+import argparse
+import asyncio
+import logging
+
+from amqp_fabric.amq_broker_connector import AmqBrokerConnector
+
+from data_agent import __version__
+from data_agent.broker.instr_exec import cli_exec
+from data_agent.broker.instr_list_services import cli_list_services
+from data_agent.config_manager import component_config_view, init_configuration
+from data_agent.config_template import CONFIG_SECTION_BROKER, CONFIG_SECTION_SERVICE
+
+# logging.basicConfig(level=os.environ.get("LOGLEVEL", "DEBUG"))
+# logging.getLogger().addHandler(logging.StreamHandler())
+
+log = logging.getLogger(__name__)
+
+
+async def create_amq_broker_connector(config, keep_alive_listen):
+    config_broker = component_config_view(config, CONFIG_SECTION_BROKER)
+    config_service = component_config_view(config, CONFIG_SECTION_SERVICE)
+
+    amq = AmqBrokerConnector(
+        amqp_uri=config_broker.uri,
+        service_domain=config_service.domain,
+        service_id=config_service.id,
+        service_type=config_service.type,
+        keep_alive_listen=keep_alive_listen,
+    )
+    await amq.open(timeout=config_broker.timeout)
+    return amq
+
+
+def run():
+    loop = asyncio.get_event_loop()
+
+    parser = argparse.ArgumentParser(description="Broker Connected Data Agent CLI")
+    subparsers = parser.add_subparsers(dest="instruction", help="Execute API call")
+    subparsers.add_parser("exec")
+    subparsers.add_parser("list_services")
+
+    config, unrecognized_args = init_configuration(
+        loop=loop, is_service=False, parser=parser
+    )
+    known_args, _ = parser.parse_known_args()
+
+    config_service = component_config_view(config, CONFIG_SECTION_SERVICE)
+
+    log.info(
+        '***** Broker Connected Agent CLI: "{}", Instruction: "{}", Ver: {} ******'.format(
+            config_service["id"], known_args.instruction, __version__
+        )
+    )
+
+    broker = loop.run_until_complete(
+        create_amq_broker_connector(
+            config=config, keep_alive_listen=(known_args.instruction == "list_services")
+        )
+    )
+
+    task = None
+
+    if known_args.instruction == "exec":
+        task = cli_exec(broker, config, unrecognized_args)
+    elif known_args.instruction == "list_services":
+        task = cli_list_services(broker, config, unrecognized_args)
+    else:
+        raise Exception("Unknown instruction")
+
+    loop.run_until_complete(task)
+
+    loop.run_until_complete(broker.close())
+    loop.run_until_complete(loop.shutdown_asyncgens())
+    loop.close()
+
+
+if __name__ == "__main__":
+    run()
diff --git a/src/data_agent/broker/instr_exec.py b/src/data_agent/broker/instr_exec.py
@@ -0,0 +1,105 @@
+import ast
+import pprint
+import time
+
+import six
+from aio_pika.exceptions import DeliveryError
+from async_timeout import timeout
+
+from data_agent.config_manager import component_config_view
+from data_agent.config_template import CONFIG_SECTION_SERVICE
+
+global_kwargs = []
+
+
+async def run_command(command, args, proxy):
+    start_time = time.time()
+
+    res = await proxy.func(method_name=command, kwargs=build_kwargs(args), expiration=3)
+
+    elapsed_time = time.time() - start_time
+    return res, elapsed_time
+
+
+def build_kwargs(args):
+    args = {i.split("=")[0]: i.split("=")[1] for i in args}
+
+    ret = {
+        k[2:]: args[k]
+        for k in args
+        if k.startswith("--") and k[2:] not in global_kwargs and args[k] is not None
+    }
+
+    # Convert types
+    for k in ret:
+        if not isinstance(ret[k], six.string_types):
+            continue
+
+        try:
+            ret[k] = ast.literal_eval(ret[k])
+
+        except ValueError:
+            pass
+        except SyntaxError:
+            pass
+
+    return ret
+
+
+async def print_formatted_result(
+    command, args, broker, service_domain, service_id, service_type
+):
+    print(f"Domain: {service_domain}")
+    print(f"Service Type: {service_type}")
+    print(f"Service Id: {service_id}")
+    print(f"Executing command: {command}")
+    # print('      args:{0}'.format(args['<args>']))
+    print(f"    kwargs:{build_kwargs(args)}")
+    print("------------------------------------------------\n\r")
+
+    proxy = await broker.rpc_proxy(service_domain, service_id, service_type)
+
+    try:
+        async with timeout(2) as cm:
+            result, elapsed_time = await run_command(command, args, proxy)
+
+        if cm.expired:
+            print("\n\r------------------------------------------------")
+            print('Error: Timeout executing command."')
+            return
+
+        if isinstance(result, list):
+            for n in result:
+                print(n)
+        else:
+            pprint.pprint(result)
+
+        print("\n\r------------------------------------------------")
+        print(
+            'Command "{0}" executed successfully ({1:.2f} sec)!'.format(
+                command, elapsed_time
+            )
+        )
+
+    except DeliveryError:
+        print(
+            "Error: Unrecognized target domain / service_id / service_type / function."
+        )
+        print("\n\r------------------------------------------------")
+
+
+async def cli_exec(broker, config, args):
+    config_service = component_config_view(config, CONFIG_SECTION_SERVICE)
+
+    service_id = config_service.id
+    service_domain = config_service.domain
+    service_type = config_service.type
+
+    await print_formatted_result(
+        args[0],
+        args[1:] if len(args) > 1 else [],
+        broker,
+        service_domain,
+        service_id,
+        service_type,
+    )
diff --git a/src/data_agent/broker/instr_list_services.py b/src/data_agent/broker/instr_list_services.py
@@ -0,0 +1,13 @@
+import asyncio
+from pprint import pprint
+
+
+async def cli_list_services(broker, config, args):
+    # Sleep to collect keep alives
+    print("Waiting 5 seconds for keep alives...")
+    await asyncio.sleep(5)
+
+    res = broker.list_services(health_check=True)
+
+    print("Detected services:")
+    pprint(res)
diff --git a/src/data_agent/connectors/fake_connector.py b/src/data_agent/connectors/fake_connector.py
@@ -1,5 +1,5 @@
+import datetime as dt
 import random
-from datetime import datetime
 from typing import Union
 
 import numpy as np
@@ -81,11 +81,11 @@ def __init__(self, conn_name="fake_client", **kwargs):
 
     def _update_random(self):
         self._tags["Random"]["Real8"]["Value"] = random.random() * 1000
-        self._tags["Random"]["Real8"]["Timestamp"] = str(datetime.utcnow())
+        self._tags["Random"]["Real8"]["Timestamp"] = str(dt.datetime.now(dt.UTC))
         self._tags["Random"]["String"]["Value"] = random.choice(
             "We are going to win this race.".split(" ")
         )
-        self._tags["Random"]["String"]["Timestamp"] = str(datetime.utcnow())
+        self._tags["Random"]["String"]["Timestamp"] = str(dt.datetime.now(dt.UTC))
 
     @property
     def connected(self):
diff --git a/src/data_agent/daq_scheduler.py b/src/data_agent/daq_scheduler.py
diff --git a/src/data_agent/local_cli.py b/src/data_agent/local_cli.py
diff --git a/src/data_agent/win32/service.py b/src/data_agent/win32/service.py
diff --git a/src/data_agent/win32/win_logger.py b/src/data_agent/win32/win_logger.py
diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml
