pb stuff

Author: gonzalocasas

.github/workflows/build.yml

@@ -40,6 +40,9 @@ jobs:
           python: '3.11'
           invoke_lint: false
           invoke_test: false
+      - name: Install test dependencies
+        run: |
+          pip install -r tests/integration/requirements.txt
       - name: Run integration tests
         run: |
           pytest tests/integration

pyproject.toml

@@ -1,3 +1,54 @@
+[build-system]
+requires = ["setuptools>=66.0"]
+build-backend = "setuptools.build_meta"
+
+# ============================================================================
+# project info
+# ============================================================================
+
+[project]
+name = "compas_eve"
+description = "COMPAS Event Extensions: adds event-based communication infrastructure to the COMPAS framework."
+keywords = ["events", "event-driven", "compas", "architecture", "distributed systems"]
+authors = [
+    { name = "Gonzalo Casas", email = "[email protected]" },
+    { name = "Chen Kasirer", email = "[email protected]" },
+]
+license = { file = "LICENSE" }
+readme = "README.md"
+requires-python = ">=3.9"
+dynamic = ['dependencies', 'optional-dependencies', 'version']
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Topic :: Scientific/Engineering",
+    "Operating System :: Unix",
+    "Operating System :: POSIX",
+    "Operating System :: Microsoft :: Windows",
+    "Programming Language :: Python",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
+]
+
+[project.entry-points.'compas_pb.plugins']
+serializers = 'compas_eve.codecs.conversions'
+
+[project.urls]
+Homepage = "https://github.com/compas-dev/compas_eve"
+Documentation = "https://compas.dev/compas_eve"
+Repository = "https://github.com/compas-dev/compas_eve"
+Changelog = "https://github.com/compas-dev/compas_eve/blob/main/CHANGELOG.md"
+Issues = "https://github.com/compas-dev/compas_eve/issues"
+Forum = "https://forum.compas-framework.org/"
+
+
+# ============================================================================
+# setuptools config
+# ============================================================================
+
 [tool.black]
 line-length = 120
 

requirements-dev.txt

@@ -3,7 +3,8 @@ autopep8
 black
 bump2version >=1.0.1
 check-manifest >=0.36
-compas_invocations
+compas_invocations2
+compas_pb >= 0.4.4
 doc8
 flake8
 invoke >=0.14

src/compas_eve/codecs.py renamed to src/compas_eve/codecs/__init__.py

@@ -88,21 +88,20 @@ def encode(self, message):
         """
         # Extract data from the message
         try:
-            data = message.data
+            return json_dumps(message.data)
         except (KeyError, AttributeError):
             try:
-                data = message.__data__
+                return json_dumps(message)
             except (KeyError, AttributeError):
-                data = dict(message)
-        return json_dumps(data)
+                return json_dumps(dict(message))
 
     def decode(self, encoded_data, message_type):
-        """Decode JSON string to message object.
+        """Decode JSON message payloads to message object.
 
         Parameters
         ----------
-        encoded_data : str
-            JSON string to decode.
+        encoded_data : bytes
+            Message bytes to decode into a JSON string.
         message_type : type
             The message type class to use for parsing.
 
@@ -111,8 +110,11 @@ def decode(self, encoded_data, message_type):
         :class:`Message`
             Decoded message object.
         """
-        data = json_loads(encoded_data)
-        return message_type.parse(data)
+        data = json_loads(encoded_data.decode())
+        if hasattr(data, "__data__"):
+            return data
+        else:
+            return message_type.parse(data)
 
 
 try:

src/compas_eve/codecs/conversions.py

@@ -0,0 +1,22 @@
+from compas_pb.core import _deserialize_any
+from compas_pb.core import _serializer_any
+from compas_pb.registry import pb_deserializer
+from compas_pb.registry import pb_serializer
+
+from compas_eve.proto import message_pb2
+from compas_eve import Message
+
+
+@pb_serializer(Message)
+def message_to_pb(message: Message) -> message_pb2.Message:
+    pb = message_pb2.Message()
+    for k, v in message.data.items():
+        pb.data[k].CopyFrom(_serializer_any(v))
+    return pb
+
+@pb_deserializer(message_pb2.Message)
+def message_from_pb(pb: message_pb2.Message) -> Message:
+    message = Message()
+    for k, v in pb.data.items():
+        message[k] = _deserialize_any(v)
+    return message

src/compas_eve/memory/__init__.py

@@ -56,7 +56,9 @@ def publish(self, topic, message):
         event_key = "event:{}".format(topic.name)
 
         def _callback(**kwargs):
-            self.emit(event_key, message)
+            encoded_message = self.codec.encode(message)
+            encoded_message_bytes = encoded_message if isinstance(encoded_message, bytes) else encoded_message.encode('utf-8')
+            self.emit(event_key, encoded_message_bytes)
 
         self.on_ready(_callback)
 
@@ -81,10 +83,14 @@ def subscribe(self, topic, callback):
         event_key = "event:{}".format(topic.name)
         subscribe_id = "{}:{}".format(event_key, id(callback))
 
+        def _local_callback(msg):
+            message_obj = self.codec.decode(msg, topic.message_type)
+            callback(message_obj)
+
         def _callback(**kwargs):
-            self.on(event_key, callback)
+            self.on(event_key, _local_callback)
 
-        self._local_callbacks[subscribe_id] = callback
+        self._local_callbacks[subscribe_id] = _local_callback
 
         self.on_ready(_callback)
 

src/compas_eve/mqtt/mqtt_paho.py

@@ -106,7 +106,7 @@ def subscribe(self, topic, callback):
         subscribe_id = "{}:{}".format(event_key, id(callback))
 
         def _local_callback(msg):
-            message_obj = self.codec.decode(msg.payload.decode(), topic.message_type)
+            message_obj = self.codec.decode(msg.payload, topic.message_type)
             callback(message_obj)
 
         def _subscribe_callback(**kwargs):

src/compas_eve/proto/message.proto

@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+package compas_eve;
+
+import "compas_pb/generated/message.proto";
+
+message Message {
+  map<string, compas_pb.data.AnyData> data = 1;
+}

src/compas_eve/proto/message_pb2.py

@@ -1,13 +1,17 @@
 from __future__ import print_function
 
 import os
+from pathlib import Path
 
-from compas_invocations import build
-from compas_invocations import docs
-from compas_invocations import style
-from compas_invocations import tests
+from compas_invocations2 import build
+from compas_invocations2 import docs
+from compas_invocations2 import style
+from compas_invocations2 import tests
 from invoke import Collection
 
+import compas_pb
+from compas_pb.invocations import generate_proto_classes
+
 ns = Collection(
     docs.help,
     style.check,
@@ -21,6 +25,7 @@
     build.prepare_changelog,
     build.clean,
     build.release,
+    generate_proto_classes,
 )
 ns.configure(
     {
@@ -30,5 +35,8 @@
             "source_dir": "src/compas_eve/ghpython/components",
             "target_dir": "src/compas_eve/ghpython/components/ghuser",
         },
+        "proto_folder": Path("./src") / "compas_eve" / "proto",
+        "proto_include_paths": [Path("./src") / "compas_eve" / "proto", compas_pb.PROTOBUF_DEFS],
+        "proto_out_folder": Path("./src") / "compas_eve" / "proto",
     }
 )