Set self.dtype only once and explain history.

src/bluesky/consolidators.py

@@ -89,10 +89,22 @@ def __init__(self, stream_resource: StreamResource, descriptor: EventDescriptor)
         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))
-        self.dtype = np.dtype(data_desc.get("dtype_numpy", 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)