Replacing AIOHTTP with aiodistbus

.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/_logger.py

@@ -76,7 +76,7 @@ def from_dict(cls, d: Dict[str, Any]):
 
 LOGGING_CONFIG: Dict[str, Any] = {
     "version": 1,
-    "disable_existing_loggers": True,
+    "disable_existing_loggers": False,
     "formatters": {
         "standard": {
             "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",

chimerapy/engine/config.py

@@ -3,6 +3,14 @@
 import pathlib
 from typing import Any, Dict
 
+# As https://github.com/lidatong/dataclasses-json/issues/202#issuecomment-1186373078
+import dataclasses_json.cfg
+
+dataclasses_json.cfg.global_config.encoders[pathlib.Path] = str
+dataclasses_json.cfg.global_config.decoders[
+    pathlib.Path
+] = pathlib.Path  # is this necessary?
+
 # Third-party Imports
 import yaml
 

chimerapy/engine/data_protocols.py

@@ -1,9 +1,20 @@
 import datetime
+import enum
+import logging
+import pathlib
+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 .node.node_config import NodeConfig
+    from .states import NodeState
 
 from dataclasses_json import DataClassJsonMixin
 
+from .networking import DataChunk
+
 
 @dataclass
 class NodePubEntry(DataClassJsonMixin):
@@ -18,6 +29,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 +38,96 @@ 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], str]
+
+
+@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
+
+
+@dataclass
+class WorkerInfo(DataClassJsonMixin):
+    host: str
+    port: int
+    logdir: pathlib.Path
+    node_config: "NodeConfig"
+    config: Dict[str, Any] = field(default_factory=dict)
+    logging_level: int = logging.INFO
+    worker_logging_port: Optional[int] = 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)