Improve custom type system

.github/workflows/python-package.yml

@@ -70,7 +70,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=runtimepy version=2.0.0
+                            repo=runtimepy version=2.1.0
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.2
-    hash=31bf81f7cf74ce38be9e121deda89fbe
+    hash=81f6301e885d2982710c34e69c537077
     =====================================
 -->
 
-# runtimepy ([2.0.0](https://pypi.org/project/runtimepy/))
+# runtimepy ([2.1.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)

local/configs/package.yaml

@@ -5,7 +5,7 @@ description: A framework for implementing Python services.
 entry: {{entry}}
 
 requirements:
-  - vcorelib>=2.0.2
+  - vcorelib>=2.5.4
   - websockets
   - "windows-curses; sys_platform == 'win32'"
 

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 2
-minor: 0
+minor: 1
 patch: 0
 entry: runtimepy

pyproject.toml

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

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.2
-# hash=25c862b538abb92065ece95c39caae77
+# hash=e2256d2409dc864fe3a4f05643274ad3
 # =====================================
 
 """
@@ -10,4 +10,4 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "2.0.0"
+VERSION = "2.1.0"

runtimepy/codec/protocol/base.py

@@ -182,10 +182,15 @@ def value(self, name: str, resolve_enum: bool = True) -> ProtocolPrimitive:
 
         return self._fields.get(name, resolve_enum=resolve_enum)
 
+    @property
+    def size(self) -> int:
+        """Get this protocol's size in bytes."""
+        return self.array.length()
+
     def __str__(self) -> str:
         """Get this instance as a string."""
 
-        return f"({self.array.size})\t" + "\t".join(
+        return f"({self.size})\t" + "\t".join(
             f"{name}={self[name]}" for name in self._names.registered_order
         )
 

runtimepy/codec/system/__init__.py

@@ -3,62 +3,112 @@
 """
 
 # built-in
-from typing import Dict, Optional
+from typing import Dict, Optional, Type
 
 # third-party
 from vcorelib.namespace import CPP_DELIM, Namespace
 
 # internal
+from runtimepy import PKG_NAME
 from runtimepy.codec.protocol import Protocol
-from runtimepy.enum import RuntimeEnum
-from runtimepy.enum.registry import EnumRegistry
+from runtimepy.enum.registry import (
+    DEFAULT_ENUM_PRIMITIVE,
+    EnumRegistry,
+    RuntimeIntEnum,
+)
 from runtimepy.primitives.byte_order import ByteOrder
 from runtimepy.primitives.type import AnyPrimitiveType, PrimitiveTypes
 
 
-class CustomType:
-    """TODO."""
-
-    def __init__(self, protocol: Protocol) -> None:
-        """Initialize this instance."""
-
-        self.protocol = protocol
-
-
 class TypeSystem:
     """A class for managing a custom type system."""
 
     def __init__(self, *namespace: str) -> None:
         """Initialize this instance."""
 
         self.primitives: Dict[str, AnyPrimitiveType] = {}
-        self.custom: Dict[str, CustomType] = {}
+        self.custom: Dict[str, Protocol] = {}
+        self._enums = EnumRegistry()
 
         global_namespace = Namespace(delim=CPP_DELIM)
 
         # Register global names.
         for name, kind in PrimitiveTypes.items():
             self.primitives[global_namespace.namespace(name)] = kind
 
-        self.root_namespace = global_namespace.child(*namespace)
+        self.root_namespace = global_namespace
 
         # Register enums.
-        self._enums = EnumRegistry()
-        self.runtime_enum(
-            "ByteOrder", ByteOrder.register_enum(self._enums, name="ByteOrder")
-        )
+        with self.root_namespace.pushed(PKG_NAME):
+            for enum in [ByteOrder]:
+                self.runtime_int_enum(enum)
+
+        self.root_namespace = global_namespace.child(*namespace)
 
-    def register(self, name: str) -> CustomType:
+    def runtime_int_enum(self, enum: Type[RuntimeIntEnum]) -> None:
+        """Register an enumeration class."""
+
+        name = self._name(enum.enum_name(), check_available=True)
+        runtime = enum.register_enum(self._enums, name=name)
+        self._register_primitive(name, runtime.primitive)
+
+    def enum(
+        self,
+        name: str,
+        items: Dict[str, int],
+        primitive: str = DEFAULT_ENUM_PRIMITIVE,
+    ) -> None:
+        """Register an enumeration."""
+
+        name = self._name(name, check_available=True)
+
+        enum = self._enums.enum(name, "int", items=items, primitive=primitive)
+        assert enum is not None
+        self._register_primitive(name, enum.primitive)
+
+    def register(self, name: str) -> Protocol:
         """Register a custom type."""
 
-        new_type = CustomType(Protocol(self._enums))
+        new_type = Protocol(self._enums)
         self.custom[self._name(name, check_available=True)] = new_type
         return new_type
 
-    def _find_name(self, name: str, strict: bool = False) -> Optional[str]:
+    def add(self, custom_type: str, field_name: str, field_type: str) -> None:
+        """Add a field to a custom type."""
+
+        type_name = self._find_name(custom_type, strict=True)
+        assert type_name is not None
+        field_type_name = self._find_name(field_type, strict=True)
+        assert field_type_name is not None
+
+        assert type_name in self.custom, type_name
+        custom = self.custom[type_name]
+
+        # Handle enumerations.
+        enum = self._enums.get(field_type_name)
+        if enum is not None:
+            custom.add_field(field_name, enum=field_type_name)
+            return
+
+        # Lookup field type.
+        if field_type_name in self.custom:
+            custom.array.add_to_end(self.custom[field_type_name].array)
+            return
+
+        custom.add_field(
+            field_name, kind=self.primitives[field_type_name].name
+        )
+
+    def _find_name(
+        self, name: str, *namespace: str, strict: bool = False
+    ) -> Optional[str]:
         """Attempt to find a registered name."""
 
-        matches = list(self.root_namespace.search(pattern=name))
+        if name in self.primitives:
+            return name
+
+        with self.root_namespace.pushed(*namespace):
+            matches = list(self.root_namespace.search(pattern=name))
 
         assert (
             0 <= len(matches) <= 1
@@ -68,50 +118,28 @@ def _find_name(self, name: str, strict: bool = False) -> Optional[str]:
 
         return matches[0] if matches else None
 
-    def _name(self, name: str, check_available: bool = False) -> str:
+    def _name(
+        self, name: str, *namespace: str, check_available: bool = False
+    ) -> str:
         """Resolve a given name against the current namespace."""
 
-        if check_available:
-            resolved = self._find_name(name)
-            assert (
-                resolved is None
-            ), f"Name '{name}' not available! found '{resolved}'"
+        with self.root_namespace.pushed(*namespace):
+            if check_available:
+                resolved = self._find_name(name)
+                assert (
+                    resolved is None
+                ), f"Name '{name}' not available! found '{resolved}'"
 
-        return self.root_namespace.namespace(name)
+            result = self.root_namespace.namespace(name)
+
+        return result
 
     def _register_primitive(self, name: str, kind: str) -> None:
-        """TODO."""
+        """Register a type alias for a primitive value."""
 
         assert name not in self.primitives, name
         self.primitives[name] = PrimitiveTypes[kind]
 
-    def runtime_enum(self, name: str, enum: RuntimeEnum) -> bool:
-        """Register an enumeration."""
-
-        name = self._name(name, check_available=True)
-
-        result = self._enums.register(name, enum)
-
-        if result:
-            self._register_primitive(name, enum.primitive)
-
-        return result
-
-    def enum(
-        self, name: str, items: Dict[str, int], primitive: str = "uint8"
-    ) -> None:
-        """Register an enumeration."""
-
-        name = self._name(name, check_available=True)
-
-        enum = self._enums.enum(name, "int", items=items, primitive=primitive)
-        assert enum is not None
-
-        # should this call "runtime_enum" ?
-        # self.runtime_enum(name, )
-
-        self._register_primitive(name, enum.primitive)
-
     def size(self, name: str) -> int:
         """Get the size of a named type."""
 
@@ -121,4 +149,4 @@ def size(self, name: str) -> int:
         if found in self.primitives:
             return self.primitives[found].size
 
-        return self.custom[found].protocol.array.length()
+        return self.custom[found].size

runtimepy/data/schemas/EnumRegistry.yaml

@@ -3,5 +3,5 @@ type: object
 additionalProperties: false
 
 patternProperties:
-  "^\\w+$":
+  "^[\\w\\:]+$":
     $ref: package://runtimepy/schemas/RuntimeEnum.yaml

runtimepy/enum/registry.py

@@ -17,6 +17,8 @@
 from runtimepy.mapping import EnumMappingData as _EnumMappingData
 from runtimepy.registry import Registry as _Registry
 
+DEFAULT_ENUM_PRIMITIVE = "uint8"
+
 
 class EnumRegistry(_Registry[_RuntimeEnum]):
     """A runtime enumeration registry."""
@@ -31,7 +33,7 @@ def enum(
         name: str,
         kind: _EnumTypelike,
         items: _EnumMappingData = None,
-        primitive: str = "uint8",
+        primitive: str = DEFAULT_ENUM_PRIMITIVE,
     ) -> _Optional[_RuntimeEnum]:
         """Create a new runtime enumeration."""
 
@@ -44,6 +46,16 @@ def enum(
 class RuntimeIntEnum(_IntEnum):
     """An integer enumeration extension."""
 
+    @classmethod
+    def primitive(cls) -> str:
+        """The underlying primitive type for this runtime enumeration."""
+        return DEFAULT_ENUM_PRIMITIVE
+
+    @classmethod
+    def enum_name(cls) -> str:
+        """Get a name for this enumeration."""
+        return cls.__name__
+
     @classmethod
     def runtime_enum(cls, identifier: int) -> _RuntimeEnum:
         """Obtain a runtime enumeration from this class."""
@@ -56,8 +68,10 @@ def register_enum(
         """Register an enumeration to a registry."""
 
         if name is None:
-            name = cls.__name__
+            name = cls.enum_name()
 
-        result = registry.register_dict(name, _RuntimeEnum.data_from_enum(cls))
+        data = _RuntimeEnum.data_from_enum(cls)
+        data["primitive"] = cls.primitive()
+        result = registry.register_dict(name, data)
         assert result is not None
         return result

runtimepy/mixins/regex.py

@@ -7,7 +7,7 @@
 from re import Pattern as _Pattern
 from re import compile as _compile
 
-DEFAULT_PATTERN = _compile("^\\w+$")
+DEFAULT_PATTERN = _compile("^[\\w\\:]+$")
 CHANNEL_PATTERN = _compile("^[a-z0-9-_.]+$")
 
 

runtimepy/requirements.txt

@@ -1,3 +1,3 @@
-vcorelib>=2.0.2
+vcorelib>=2.5.4
 websockets
 windows-curses; sys_platform == 'win32'