SharedMemory and ndarray serialization

dev-environment.yml

@@ -18,6 +18,7 @@ name: appose-dev
 channels:
   - conda-forge
   - defaults
+  - forklift
 dependencies:
   - python >= 3.10
   # Developer tools

src/appose/types.py

@@ -28,6 +28,7 @@
 ###
 
 import json
+from multiprocessing import shared_memory
 from typing import Any, Dict
 
 Args = Dict[str, Any]
@@ -38,4 +39,42 @@ def encode(data: Args) -> str:
 
 
 def decode(the_json: str) -> Args:
-    return json.loads(the_json)
+    return json.loads(the_json, object_hook=_appose_object_hook)
+
+
+class ShmNDArray:
+    def __init__(self, shm: shared_memory.SharedMemory, dtype: str, shape):
+        self.shm = shm
+        self.dtype = dtype
+        self.shape = shape
+
+    def __str__(self):
+        return (
+            f"ShmNDArray("
+            f"shm='{self.shm.name}' ({self.shm.size}), "
+            f"dtype='{self.dtype}', "
+            f"shape={self.shape})"
+        )
+
+    def ndarray(self):
+        try:
+            import math
+
+            import numpy
+
+            num_elements = math.prod(self.shape)
+            return numpy.ndarray(
+                num_elements, dtype=self.dtype, buffer=self.shm.buf
+            ).reshape(self.shape)
+        except ModuleNotFoundError:
+            raise ImportError("NumPy is not available.")
+
+
+def _appose_object_hook(obj: Dict):
+    type = obj.get("appose_type")
+    if type == "shm":
+        return shared_memory.SharedMemory(name=(obj["name"]), size=(obj["size"]))
+    elif type == "ndarray":
+        return ShmNDArray(obj["shm"], obj["dtype"], obj["shape"])
+    else:
+        return obj