Dataclass

bigraph_schema/data.py

@@ -0,0 +1 @@
+

bigraph_schema/registry.py

@@ -13,9 +13,12 @@
 import numpy as np
 
 from pprint import pformat as pf
+from typing import Any, Tuple, Union, Optional, Mapping
+from dataclasses import field, make_dataclass
 
 from bigraph_schema.parse import parse_expression
 from bigraph_schema.protocols import local_lookup_module, function_module
+import bigraph_schema.data
 
 
 NONE_SYMBOL = '!nil'
@@ -525,6 +528,34 @@ def fold_union(schema, state, method, values, core):
     return result
 
 
+def type_parameters_for(schema):
+    parameters = []
+    for key in schema['_type_parameters']:
+        subschema = schema.get(f'_{key}', 'any')
+        parameters.append(subschema)
+
+    return parameters
+
+
+def dataclass_union(schema, path, core):
+    parameters = type_parameters_for(schema)
+    subtypes = []
+    for parameter in parameters:
+        dataclass = core.dataclass(
+            parameter,
+            path)
+
+        if isinstance(dataclass, str):
+            subtypes.append(dataclass)
+        elif isinstance(dataclass, type):
+            subtypes.append(dataclass.__name__)
+        else:
+            subtypes.append(str(dataclass))
+
+    parameter_block = ', '.join(subtypes)
+    return eval(f'Union[{parameter_block}]')
+
+
 def divide_any(schema, state, values, core):
     divisions = values.get('divisions', 2)
 
@@ -551,6 +582,64 @@ def divide_any(schema, state, values, core):
             for _ in range(divisions)]
 
 
+def dataclass_any(schema, path, core):
+    parts = path
+    if not parts:
+        parts = ['top']
+    dataclass_name = '_'.join(parts)
+
+    if isinstance(schema, dict):
+        type_name = schema.get('_type', 'any')
+
+        branches = {}
+        for key, subschema in schema.items():
+            if not key.startswith('_'):
+                branch = core.dataclass(
+                    subschema,
+                    path + [key])
+
+                def default(subschema=subschema):
+                    return core.default(subschema)
+
+                branches[key] = (
+                    key,
+                    branch,
+                    field(default_factory=default))
+
+        dataclass = make_dataclass(
+            dataclass_name,
+            branches.values(),
+            namespace={
+                '__module__': 'bigraph_schema.data'})
+
+        setattr(
+            bigraph_schema.data,
+            dataclass_name,
+            dataclass)
+
+    else:
+        schema = core.access(schema)
+        dataclass = core.dataclass(schema, path)
+
+    return dataclass
+
+
+def dataclass_tuple(schema, path, core):
+    parameters = type_parameters_for(schema)
+    subtypes = []
+
+    for index, key in enumerate(schema['type_parameters']):
+        subschema = schema.get(key, 'any')
+        subtype = core.dataclass(
+            subschema,
+            path + [index])
+
+        subtypes.append(subtype)
+
+    parameter_block = ', '.join(subtypes)
+    return eval(f'tuple[{parameter_block}]')
+
+
 def divide_tuple(schema, state, values, core):
     divisions = values.get('divisions', 2)
 
@@ -954,6 +1043,7 @@ def deserialize_union(schema, encoded, core):
         '_check': check_any,
         '_serialize': serialize_any,
         '_deserialize': deserialize_any,
+        '_dataclass': dataclass_any,
         '_fold': fold_any,
         '_merge': merge_any,
         '_bind': bind_any,
@@ -967,6 +1057,7 @@ def deserialize_union(schema, encoded, core):
         '_slice': slice_tuple,
         '_serialize': serialize_tuple,
         '_deserialize': deserialize_tuple,
+        '_dataclass': dataclass_tuple,
         '_fold': fold_tuple,
         '_divide': divide_tuple,
         '_bind': bind_tuple,
@@ -980,6 +1071,7 @@ def deserialize_union(schema, encoded, core):
         '_slice': slice_union,
         '_serialize': serialize_union,
         '_deserialize': deserialize_union,
+        '_dataclass': dataclass_union,
         '_fold': fold_union,
         '_description': 'union of a set of possible types'}}