Merge pull request bluesky#1796 from jwlodek/use-dtype-numpy

Update to allow for using dtype_numpy key alongside old dtype_str

src/bluesky/consolidators.py

@@ -85,13 +85,26 @@ def __init__(self, stream_resource: StreamResource, descriptor: EventDescriptor)
         self.assets: List[Asset] = []
         self._sres_parameters = stream_resource["parameters"]
 
-        # Find data shape and machine dtype; dtype_str takes precedence if specified
+        # Find data shape and machine dtype; dtype_numpy, dtype_str take precedence if specified
         data_desc = descriptor["data_keys"][self.data_key]
         self.datum_shape = tuple(data_desc["shape"])
         self.datum_shape = self.datum_shape if self.datum_shape != (1,) else ()
-        self.dtype = data_desc["dtype"]
-        self.dtype = DTYPE_LOOKUP[self.dtype] if self.dtype in DTYPE_LOOKUP.keys() else self.dtype
-        self.dtype = np.dtype(data_desc.get("dtype_str", self.dtype))
+        # Get data type. From highest precedent to lowest:
+        # 1. Try 'dtype_numpy', optional in the document schema.
+        # 2. Try 'dtype_str', an old convention predataing 'dtype_numpy', not in the schema.
+        # 3. Get 'dtype', required by the schema, which is a fuzzy JSON spec like 'number'
+        #    and make a best effort to convert it to a numpy spec like '<u8'.
+        # 4. If unable to do any of the above, pass through whatever string is in 'dtype'.
+        self.dtype = (
+            np.dtype(
+                data_desc.get("dtype_numpy")  # standard location
+                or data_desc.get(
+                    "dtype_str",  # legacy location
+                    # try to guess numpy dtype from JSON type
+                    DTYPE_LOOKUP.get(data_desc["dtype"], data_desc["dtype"])
+                )
+            )
+        )
         self.chunk_size = self._sres_parameters.get("chunk_size", None)
 
         self._num_rows: int = 0  # Number of rows in the Data Source (all rows, includung skips)

src/bluesky/tests/test_consolidators.py

@@ -10,7 +10,7 @@ def descriptor():
             "test_img": {
                 "shape": [10, 15],
                 "dtype": "array",
-                "dtype_str": "<f8",
+                "dtype_numpy": "<f8",
                 "external": "STREAM:",
                 "object_name": "test_object",
             },

src/bluesky/tests/test_tiled_writer.py

@@ -149,7 +149,7 @@ def describe(self) -> Dict[str, DataKey]:
             f"{self.name}-sd1": DataKey(source="file", dtype="number", shape=[1], external="STREAM:"),
             f"{self.name}-sd2": DataKey(source="file", dtype="array", shape=[10, 15], external="STREAM:"),
             f"{self.name}-sd3": DataKey(
-                source="file", dtype="array", dtype_str="uint8", shape=[5, 7, 4], external="STREAM:"
+                source="file", dtype="array", dtype_numpy="uint8", shape=[5, 7, 4], external="STREAM:"
             ),
         }
 
@@ -188,8 +188,8 @@ class SynSignalWithRegistry(ophyd.sim.SynSignalWithRegistry):
     Subclassed from ophyd.sim to match the updated schema of Resource documents.
     """
 
-    def __init__(self, *args, dtype_str="uint8", **kwargs):
-        self.dtype_str = dtype_str
+    def __init__(self, *args, dtype_numpy="uint8", **kwargs):
+        self.dtype_numpy = dtype_numpy
         super().__init__(*args, **kwargs)
 
     def stage(self):
@@ -201,7 +201,7 @@ def describe(self):
         res = super().describe()
         for key in res:
             res[key]["external"] = "FILESTORE"
-            res[key]["dtype_str"] = self.dtype_str
+            res[key]["dtype_numpy"] = self.dtype_numpy
         return res
 
 
@@ -232,7 +232,7 @@ def test_stream_datum_collectable(RE, client, tmp_path):
 def test_handling_non_stream_resource(RE, client, tmp_path):
     det = SynSignalWithRegistry(
         func=lambda: np.random.randint(0, 255, (10, 15), dtype="uint8"),
-        dtype_str="uint8",
+        dtype_numpy="uint8",
         name="img",
         labels={"detectors"},
         save_func=tf.imwrite,