Implement the NullPatternGenerator

src/ophyd_async/sim/__init__.py

@@ -9,7 +9,7 @@
     VerticalMirror,
 )
 from ._motor import FlySimMotorInfo, SimMotor
-from ._pattern_generator import PatternGenerator
+from ._pattern_generator import NullPatternGenerator, PatternGenerator
 from ._point_detector import SimPointDetector
 from ._stage import SimStage
 
@@ -18,6 +18,7 @@
     "FlySimMotorInfo",
     "SimStage",
     "PatternGenerator",
+    "NullPatternGenerator",
     "SimPointDetector",
     "SimBlobDetector",
     "VerticalMirror",

src/ophyd_async/sim/_pattern_generator.py

@@ -124,3 +124,31 @@ def close_file(self):
         if self._file:
             self._file.close()
             self._file = None
+
+
+class NullPatternGenerator(PatternGenerator):
+    """A pattern generator that generates all zeros and saves no file."""
+
+    def __init__(self, sleep=asyncio.sleep):
+        super().__init__(sleep)
+        self.n_images = 0
+
+    def open_file(self, path: Path, width: int, height: int):
+        pass
+
+    async def write_images_to_file(
+        self, exposure: float, period: float, number_of_frames: int
+    ):
+        self.n_images += number_of_frames
+
+    def generate_point(self, channel: int = 1, high_energy: bool = False) -> float:
+        return 0.0
+
+    async def wait_for_next_index(self, timeout: float):
+        pass
+
+    def get_last_index(self) -> int:
+        return self.n_images
+
+    def close_file(self):
+        pass

tests/sim/test_pattern_generator.py

@@ -0,0 +1,41 @@
+from collections import defaultdict
+
+import bluesky.plans as bp
+from bluesky import RunEngine
+
+from ophyd_async.core import StaticPathProvider
+from ophyd_async.sim import NullPatternGenerator, SimBlobDetector
+from ophyd_async.testing import assert_emitted
+
+
+async def test_null_pattern_generator_does_nothing(RE: RunEngine):
+    pattern_generator = NullPatternGenerator()
+    path_provider = StaticPathProvider(lambda _: "null_file", "/")
+    detector = SimBlobDetector(
+        path_provider, pattern_generator=pattern_generator, name="det"
+    )
+    docs = defaultdict(list)
+    RE.subscribe(lambda name, doc: docs[name].append(doc))
+    RE(bp.count([detector], num=2))
+
+    assert_emitted(
+        docs, start=1, descriptor=1, stream_resource=2, stream_datum=4, event=2, stop=1
+    )
+    assert docs["descriptor"][0]["data_keys"] == {
+        "det": {
+            "source": "sim://pattern-generator-hdf-file",
+            "shape": [1, 240, 320],
+            "dtype": "array",
+            "dtype_numpy": "|u1",
+            "object_name": "det",
+            "external": "STREAM:",
+        },
+        "det-sum": {
+            "source": "sim://pattern-generator-hdf-file",
+            "shape": [],
+            "dtype": "number",
+            "dtype_numpy": "<i8",
+            "object_name": "det",
+            "external": "STREAM:",
+        },
+    }