Merge pull request #30 from vkottler/dev/protocol-dev

Dev/protocol dev

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.0
-    hash=edfe0b431e5e5c87a6a1e84b9e3a12f6
+    hash=a8255694100a058fa4413abe0341a123
     =====================================
 -->
 
-# runtimepy ([0.12.3](https://pypi.org/project/runtimepy/))
+# runtimepy ([0.13.0](https://pypi.org/project/runtimepy/))
 
 [![python](https://img.shields.io/pypi/pyversions/runtimepy.svg)](https://pypi.org/project/runtimepy/)
 ![Build Status](https://github.com/vkottler/runtimepy/workflows/Python%20Package/badge.svg)
@@ -41,7 +41,7 @@ This package is tested on the following platforms:
 # Command-line Options
 
 ```
-$ ./venv3.7/bin/runtimepy -h
+$ ./venv3.8/bin/runtimepy -h
 
 usage: runtimepy [-h] [--version] [-v] [-C DIR] {tui,noop} ...
 

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 0
-minor: 12
-patch: 3
+minor: 13
+patch: 0
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "0.12.3"
+version = "0.13.0"
 description = "A framework for implementing Python services."
 readme = "README.md"
 requires-python = ">=3.7"

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.0
-# hash=f5cf7cc360be1ec5780bef4fa93e0a53
+# hash=1a71c8001105fcdb5a240f10db7ba967
 # =====================================
 
 """
@@ -10,4 +10,4 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "0.12.3"
+VERSION = "0.13.0"

runtimepy/codec/protocol/base.py

@@ -62,9 +62,12 @@ def __init__(
         names: _NameRegistry = None,
         fields: BitFieldsManager = None,
         build: _List[_Union[int, FieldSpec]] = None,
+        identifier: int = 1,
     ) -> None:
         """Initialize this protocol."""
 
+        self.id = identifier
+
         # Each instance gets its own array.
         self.array = PrimitiveArray()
 

runtimepy/codec/protocol/json.py

@@ -3,6 +3,8 @@
 """
 
 # built-in
+from json import dumps as _dumps
+from typing import BinaryIO as _BinaryIO
 from typing import Dict as _Dict
 from typing import List as _List
 from typing import Set as _Set
@@ -25,18 +27,32 @@
 )
 
 BUILD_KEY = "build"
+META_KEY = "meta"
 T = _TypeVar("T", bound="JsonProtocol")
 
 
 class JsonProtocol(ProtocolBase):
     """A class for defining runtime communication protocols."""
 
+    def meta_str(self, resolve_enum: bool = True) -> str:
+        """Get protocol metadata as a string.."""
+        return _dumps(self.export_json(resolve_enum=resolve_enum))
+
+    def meta_bytes(self, resolve_enum: bool = True) -> bytes:
+        """Get protocol metadata as a bytes object."""
+        return self.meta_str(resolve_enum=resolve_enum).encode()
+
+    def write_meta(self, stream: _BinaryIO, resolve_enum: bool = True) -> int:
+        """Write protocol metadata to a stream."""
+        return stream.write(self.meta_bytes(resolve_enum=resolve_enum))
+
     def export_json(
         self, resolve_enum: bool = True
     ) -> _Dict[str, _JsonObject]:
         """Export this protocol's data to JSON."""
 
         data = self._fields.export_json(resolve_enum=resolve_enum)
+        data[META_KEY] = {"id": self.id}
 
         # Export regular-field names.
         json_obj = data[NAMES_KEY]
@@ -101,7 +117,13 @@ def import_json(cls: _Type[T], data: _Dict[str, _JsonObject]) -> T:
                     )
                 )
 
-        result = cls(fields.enums, fields.registry, fields=fields, build=build)
+        result = cls(
+            fields.enums,
+            fields.registry,
+            fields=fields,
+            build=build,
+            identifier=_cast(int, data[META_KEY]["id"]),
+        )
 
         # Set values.
         for name, value in values.items():

runtimepy/enum/__init__.py

@@ -3,7 +3,9 @@
 """
 
 # built-in
+from enum import IntEnum as _IntEnum
 from typing import Optional as _Optional
+from typing import Type as _Type
 from typing import Union as _Union
 from typing import cast as _cast
 
@@ -138,3 +140,20 @@ def register_int(self, name: str) -> _Optional[int]:
     def register_bool(self, name: str, value: bool) -> bool:
         """Register a boolean enumeration."""
         return self.bools.register(name, value)
+
+    @staticmethod
+    def data_from_enum(enum: _Type[_IntEnum]) -> _JsonObject:
+        """Get JSON data from an enumeration class."""
+
+        return {
+            "type": "int",
+            "items": {x.name.lower(): x.value for x in enum},
+        }
+
+    @staticmethod
+    def from_enum(enum: _Type[_IntEnum], identifier: int) -> "RuntimeEnum":
+        """Create a runtime enumeration from an enum class."""
+
+        data = RuntimeEnum.data_from_enum(enum)
+        data["id"] = identifier
+        return RuntimeEnum.create(data)

runtimepy/enum/registry.py

@@ -3,6 +3,7 @@
 """
 
 # built-in
+from enum import IntEnum as _IntEnum
 from typing import Optional as _Optional
 from typing import Type as _Type
 from typing import cast as _cast
@@ -37,3 +38,25 @@ def enum(
         if items is not None:
             data["items"] = items  # type: ignore
         return self.register_dict(name, data)
+
+
+class RuntimeIntEnum(_IntEnum):
+    """An integer enumeration extension."""
+
+    @classmethod
+    def runtime_enum(cls, identifier: int) -> _RuntimeEnum:
+        """Obtain a runtime enumeration from this class."""
+        return _RuntimeEnum.from_enum(cls, identifier)
+
+    @classmethod
+    def register_enum(
+        cls, registry: EnumRegistry, name: str = None
+    ) -> _RuntimeEnum:
+        """Register an enumeration to a registry."""
+
+        if name is None:
+            name = cls.__name__
+
+        result = registry.register_dict(name, _RuntimeEnum.data_from_enum(cls))
+        assert result is not None
+        return result

runtimepy/telemetry/__init__.py

@@ -0,0 +1,13 @@
+"""
+A module implementing a telemetry interface.
+"""
+
+# internal
+from runtimepy.enum.registry import RuntimeIntEnum as _RuntimeIntEnum
+
+
+class MessageType(_RuntimeIntEnum):
+    """An enumeration of viable message types."""
+
+    PROTOCOL_META = 1
+    PROTOCOL_DATA = 2

tests/codec/test_protocol.py

@@ -4,6 +4,8 @@
 
 # built-in
 from copy import copy
+from io import BytesIO
+from json import load
 
 # module under test
 from runtimepy.codec.protocol import Protocol
@@ -53,3 +55,8 @@ def test_protocol_basic():
     new_proto = Protocol.import_json(proto.export_json())
     assert new_proto["test1"] == 40
     assert proto["flag1"] == "valve_open"
+
+    with BytesIO() as stream:
+        assert proto.write_meta(stream)
+        stream.seek(0)
+        assert Protocol.import_json(load(stream))

tests/enum/test_enum.py

@@ -0,0 +1,26 @@
+"""
+Test the 'enum' module.
+"""
+
+# built-in
+from enum import IntEnum, auto
+
+# module under test
+from runtimepy.enum import RuntimeEnum
+
+
+class SampleEnum(IntEnum):
+    """A sample enumeration."""
+
+    A = auto()
+    B = auto()
+    C = auto()
+
+
+def test_runtime_enum_from_enum():
+    """Test that we can create a runtime enumeration from a class."""
+
+    enum = RuntimeEnum.from_enum(SampleEnum, 1)
+    assert enum.get_int("a")
+    assert enum.get_int("b")
+    assert enum.get_int("c")

tests/telemetry/test_telemetry.py

@@ -0,0 +1,21 @@
+"""
+Test the 'telemetry' module.
+"""
+
+# module under test
+from runtimepy.enum.registry import EnumRegistry
+from runtimepy.telemetry import MessageType
+
+
+def test_message_type_basic():
+    """Test basic interactions with the message-type enumeration."""
+
+    enum = MessageType.runtime_enum(1)
+    assert enum.get_int("protocol_meta") == 1
+    assert enum.get_int("protocol_data") == 2
+
+    enum_reg = EnumRegistry({})
+    enum = MessageType.register_enum(enum_reg)
+    assert enum_reg["MessageType"] is enum
+    assert enum.get_int("protocol_meta") == 1
+    assert enum.get_int("protocol_data") == 2