Transition local eventbuses to use aiodistbus.EventBus. (#291)

* WIP: Node eventbus refactoring.

* Refactored ProcessorService

* Node services now using new aiodistbus EventBus.

* Worker's NodeHandlerService tests passing.

* Converted all local eventbus in the Worker to use aiodistbus.

* Finished Manager, Worker, and Node refactoring to use aiodistbus's local EventBus implementation.

* Reconfigured testing to improve testing speed and separation.

* Correct TEST_DATA_DIR variable after file move.

* Removed docker tests.

* Removed non-public APIs that are unnecessary.

* Refactored the Worker's services to improve decoupling.

* Finished refactoring structs into a single data_protocols file to improve model re-used.

* Fixing minor issues.

Author: edavalosanaya

.github/workflows/test.yml

@@ -67,17 +67,35 @@ jobs:
           ${{ steps.cp39.outputs.python-path }} -m pip install .[test]
           echo "MANUAL_OS_SET=Windows" >> $GITHUB_ENV
 
-      - name: Perform faster tests
+      - name: Perform ChimeraPy utils tests
         run: |
-          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m "not slow" test
+          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m test/core
           ${{ steps.cp39.outputs.python-path }} -m coverage combine --append
-          mv chimerapy-engine-test.log chimerapy-engine-test-fast.log
+          mv chimerapy-engine-test.log chimerapy-engine-test-utils.log
 
-      - name: Perform slower tests
+      - name: Perform ChimeraPy logger tests
         run: |
-          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m "slow" test
+          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m test/logger
           ${{ steps.cp39.outputs.python-path }} -m coverage combine --append
-          mv chimerapy-engine-test.log chimerapy-engine-test-slow.log
+          mv chimerapy-engine-test.log chimerapy-engine-test-logger.log
+
+      - name: Perform ChimeraPy Node tests
+        run: |
+          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m test/node
+          ${{ steps.cp39.outputs.python-path }} -m coverage combine --append
+          mv chimerapy-engine-test.log chimerapy-engine-test-node.log
+
+      - name: Perform ChimeraPy Worker tests
+        run: |
+          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m test/worker
+          ${{ steps.cp39.outputs.python-path }} -m coverage combine --append
+          mv chimerapy-engine-test.log chimerapy-engine-test-worker.log
+
+      - name: Perform ChimeraPy Manager tests
+        run: |
+          ${{ steps.cp39.outputs.python-path }} -m coverage run --source=chimerapy -m pytest -v --reruns 5 --color yes --reruns-delay 5 -m test/manager
+          ${{ steps.cp39.outputs.python-path }} -m coverage combine --append
+          mv chimerapy-engine-test.log chimerapy-engine-test-manager.log
 
       - name: Combine test logs
         run : |

chimerapy/engine/data_protocols.py

@@ -1,9 +1,18 @@
 import datetime
+import enum
+import logging
+import typing
 from dataclasses import dataclass, field
-from typing import Dict
+from typing import Any, Dict, List, Literal, Optional, Union
+
+if typing.TYPE_CHECKING:
+    from .graph import Graph
+    from .states import NodeState
 
 from dataclasses_json import DataClassJsonMixin
 
+from .networking import DataChunk
+
 
 @dataclass
 class NodePubEntry(DataClassJsonMixin):
@@ -18,6 +27,7 @@ class NodePubTable(DataClassJsonMixin):
 
 @dataclass
 class NodeDiagnostics(DataClassJsonMixin):
+    node_id: str = ""
     timestamp: str = field(
         default_factory=lambda: str(datetime.datetime.now().isoformat())
     )  # ISO str
@@ -26,3 +36,85 @@ class NodeDiagnostics(DataClassJsonMixin):
     memory_usage: float = 0  # KB
     cpu_usage: float = 0  # percentage
     num_of_steps: int = 0
+
+
+########################################################################
+## Manager specific
+########################################################################
+
+
+@dataclass
+class RegisterMethodResponseData(DataClassJsonMixin):
+    success: bool
+    result: Dict[str, Any]
+
+
+@dataclass
+class UpdateSendArchiveData(DataClassJsonMixin):
+    worker_id: str
+    success: bool
+
+
+@dataclass
+class CommitData(DataClassJsonMixin):
+    graph: "Graph"
+    mapping: Dict[str, List[str]]
+    context: Literal["multiprocessing", "threading"] = "multiprocessing"
+    send_packages: Optional[List[Dict[str, Any]]] = None
+
+
+########################################################################
+## Worker specific
+########################################################################
+
+
+@dataclass
+class ConnectData(DataClassJsonMixin):
+    method: Literal["ip", "zeroconf"]
+    host: Optional[str] = None
+    port: Optional[int] = None
+
+
+@dataclass
+class GatherData(DataClassJsonMixin):
+    node_id: str
+    output: Union[DataChunk, List[int]]
+
+
+@dataclass
+class ResultsData(DataClassJsonMixin):
+    node_id: str
+    success: bool
+    output: Any
+
+
+@dataclass
+class ServerMessage(DataClassJsonMixin):
+    signal: enum.Enum
+    data: Dict[str, Any] = field(default_factory=dict)
+    client_id: Optional[str] = None
+
+
+########################################################################
+## Node specific
+########################################################################
+
+
+@dataclass
+class PreSetupData(DataClassJsonMixin):
+    state: "NodeState"
+    logger: logging.Logger
+
+
+@dataclass
+class RegisteredMethod(DataClassJsonMixin):
+    name: str
+    style: Literal["concurrent", "blocking", "reset"] = "concurrent"
+    params: Dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class RegisteredMethodData(DataClassJsonMixin):
+    node_id: str
+    method_name: str
+    params: Dict[str, Any] = field(default_factory=dict)