docs: Add simple devices and plans for use in tutorials

src/blueapi/core/protocols.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass, field
 from typing import Any, Protocol, runtime_checkable
 
 
@@ -25,3 +26,19 @@ def build_and_connect(
         timeout: float | None = None,
         fixtures: dict[str, Any] | None = None,
     ) -> DeviceConnectResult: ...
+
+
+@dataclass
+class StaticDeviceManager:
+    devices: dict[str, Any] = field(default_factory=dict)
+    build_errors: dict[str, Exception] = field(default_factory=dict)
+    connection_errors: dict[str, Exception] = field(default_factory=dict)
+
+    def build_and_connect(
+        self,
+        *,
+        mock: bool = False,
+        timeout: float | None = None,
+        fixtures: dict[str, Any] | None = None,
+    ) -> DeviceConnectResult:
+        return self

src/blueapi/tutorial/config.yaml

@@ -0,0 +1,11 @@
+env:
+  metadata:
+    instrument: tutorial-instrument
+  sources:
+    - kind: deviceManager
+      module: blueapi.tutorial.devices
+    - kind: planFunctions
+      module: blueapi.tutorial.plans
+stomp:
+  enabled: true
+  url: tcp://localhost:61613/

src/blueapi/tutorial/devices.py

@@ -0,0 +1,11 @@
+from ophyd_async.sim import SimMotor
+
+from blueapi.core.protocols import StaticDeviceManager
+
+devices = StaticDeviceManager()
+
+x = SimMotor(name="x")
+y = SimMotor(name="y")
+
+devices.devices["x"] = x
+devices.devices["y"] = y

src/blueapi/tutorial/plans.py

@@ -0,0 +1,43 @@
+from collections.abc import Sequence
+from typing import Annotated, Any
+
+import bluesky.plans as bp
+from bluesky.protocols import Readable
+from bluesky.utils import MsgGenerator
+from ophyd_async.core import AsyncReadable
+from pydantic import Field, NonNegativeFloat, validate_call
+
+
+@validate_call(config={"arbitrary_types_allowed": True})
+def count(
+    detectors: Annotated[
+        Sequence[Readable | AsyncReadable],
+        Field(
+            description="Set of readable devices, will take a reading at each point",
+            min_length=1,
+        ),
+    ],
+    num: Annotated[int, Field(description="Number of frames to collect", ge=1)] = 1,
+    delay: Annotated[
+        NonNegativeFloat | Sequence[NonNegativeFloat],
+        Field(
+            description="Delay between readings: if tuple, len(delay) == num - 1 and \
+            the delays are between each point, if value or None is the delay for every \
+            gap",
+            json_schema_extra={"units": "s"},
+        ),
+    ] = 0.0,
+    metadata: dict[str, Any] | None = None,
+) -> MsgGenerator:
+    """Reads from a number of devices.
+
+    Wraps bluesky.plans.count(det, num, delay, md=metadata) exposing only serializable
+    parameters and metadata.
+    """
+    if isinstance(delay, Sequence):
+        assert len(delay) == num - 1, (
+            f"Number of delays given must be {num - 1}: was given {len(delay)}"
+        )
+    metadata = metadata or {}
+    metadata["shape"] = (num,)
+    yield from bp.count(tuple(detectors), num, delay=delay, md=metadata)

tests/unit_tests/core/test_context.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 from types import ModuleType, NoneType
-from typing import Any, Generic, TypeVar, Union
+from typing import Generic, TypeVar, Union
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
@@ -45,7 +45,7 @@
 )
 from blueapi.core import BlueskyContext, is_bluesky_compatible_device
 from blueapi.core.context import DefaultFactory, generic_bounds, qualified_name
-from blueapi.core.protocols import DeviceConnectResult, DeviceManager
+from blueapi.core.protocols import DeviceManager, StaticDeviceManager
 from blueapi.utils.invalid_config_error import InvalidConfigError
 
 SIM_MOTOR_NAME = "sim"
@@ -669,22 +669,6 @@ def demo_plan(foo: int | None) -> MsgGenerator:
     assert "foo" in schema.get("required", [])
 
 
-@dataclass
-class StaticDeviceManager:
-    devices: dict[str, Any] = field(default_factory=dict)
-    build_errors: dict[str, Exception] = field(default_factory=dict)
-    connection_errors: dict[str, Exception] = field(default_factory=dict)
-
-    def build_and_connect(
-        self,
-        *,
-        mock: bool = False,
-        timeout: float | None = None,
-        fixtures: dict[str, Any] | None = None,
-    ) -> DeviceConnectResult:
-        return self
-
-
 def test_empty_device_manager(empty_context: BlueskyContext):
     sdm = StaticDeviceManager()
     empty_context.with_device_manager(sdm)