Merge pull request #304 from DiamondLightSource/merge-dls-bluesky-core

Move plans, stubs and utils from dls-bluesky-core deprecated package to dodal

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
     "opencv-python-headless",                                                                          # For pin-tip detection.
     "aioca",                                                                                           # Required for CA support with ophyd-async.
     "p4p",                                                                                             # Required for PVA support with ophyd-async.
+    "numpy",
 ]
 
 dynamic = ["version"]
@@ -88,6 +89,11 @@ addopts = """
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
 
+[tool.coverage.report]
+exclude_also = [
+    '^"""', # Ignore the start/end of a file-level triple quoted docstring
+]
+
 [tool.coverage.run]
 data_file = "/tmp/dodal.coverage"
 

src/dodal/common/__init__.py

@@ -0,0 +1,12 @@
+from .coordination import group_uuid, inject
+from .maths import in_micros, step_to_num
+from .types import MsgGenerator, PlanGenerator
+
+__all__ = [
+    "group_uuid",
+    "inject",
+    "in_micros",
+    "MsgGenerator",
+    "PlanGenerator",
+    "step_to_num",
+]

src/dodal/common/coordination.py

@@ -0,0 +1,38 @@
+import uuid
+
+from dodal.common.types import Group
+
+
+def group_uuid(name: str) -> Group:
+    """
+    Returns a unique but human-readable string, to assist debugging orchestrated groups.
+
+    Args:
+        name (str): A human readable name
+
+    Returns:
+        readable_uid (Group): name appended with a unique string
+    """
+    return f"{name}-{str(uuid.uuid4())[:6]}"
+
+
+def inject(name: str):  # type: ignore
+    """
+    Function to mark a default argument of a plan method as a reference to a device
+    that is stored in the Blueapi context, as devices are constructed on startup of the
+    service, and are not available to be used when writing plans.
+    Bypasses mypy linting, returning x as Any and therefore valid as a default
+    argument.
+    e.g. For a 1-dimensional scan, that is usually performed on a consistent Movable
+    axis with name "stage_x"
+    def scan(x: Movable = inject("stage_x"), start: float = 0.0 ...)
+
+    Args:
+        name (str): Name of a device to be fetched from the Blueapi context
+
+    Returns:
+        Any: name but without typing checking, valid as any default type
+
+    """
+
+    return name

src/dodal/common/maths.py

@@ -0,0 +1,52 @@
+from typing import Tuple
+
+import numpy as np
+
+
+def step_to_num(start: float, stop: float, step: float) -> Tuple[float, float, int]:
+    """
+    Standard handling for converting from start, stop, step to start, stop, num
+    Forces step to be same direction as length
+    Includes a final point if it is within 1% of the final step, prevents floating
+    point arithmatic errors from giving inconsistent shaped scans between steps of an
+    outer axis.
+
+    Args:
+        start (float):
+            Start of length, will be returned unchanged
+        stop (float):
+            End of length, if length/step does not divide cleanly will be returned
+            extended up to 1% of step, or else truncated.
+        step (float):
+            Length of a step along the line formed from start to stop.
+            If stop < start, will be coerced to be backwards.
+
+    Returns:
+        start, adjusted_stop, num = Tuple[float, float, int]
+        start will be returned unchanged
+        adjusted_stop = start + (num - 1) * step
+        num is the maximal number of steps that could fit into the length.
+
+    """
+    # Make step be the right direction
+    step = abs(step) if stop >= start else -abs(step)
+    # If stop is within 1% of a step then include it
+    steps = int((stop - start) / step + 0.01)
+    return start, start + steps * step, steps + 1  # include 1st point
+
+
+def in_micros(t: float) -> int:
+    """
+    Converts between a positive number of seconds and an equivalent
+    number of microseconds.
+
+    Args:
+        t (float): A time in seconds
+    Raises:
+        ValueError: if t < 0
+    Returns:
+        t (int): A time in microseconds, rounded up to the nearest whole microsecond,
+    """
+    if t < 0:
+        raise ValueError(f"Expected a positive time in seconds, got {t!r}")
+    return int(np.ceil(t * 1e6))

src/dodal/common/types.py

@@ -0,0 +1,17 @@
+from typing import (
+    Annotated,
+    Any,
+    Callable,
+    Generator,
+)
+
+from bluesky.utils import Msg
+
+Group = Annotated[str, "String identifier used by 'wait' or stubs that await"]
+MsgGenerator = Annotated[
+    Generator[Msg, Any, None],
+    "A true 'plan', usually the output of a generator function",
+]
+PlanGenerator = Annotated[
+    Callable[..., MsgGenerator], "A function that generates a plan"
+]

tests/common/test_coordination.py

@@ -0,0 +1,12 @@
+import uuid
+
+import pytest
+
+from dodal.common.coordination import group_uuid
+
+
+@pytest.mark.parametrize("group", ["foo", "bar", "baz", str(uuid.uuid4())])
+def test_group_uid(group: str):
+    gid = group_uuid(group)
+    assert gid.startswith(f"{group}-")
+    assert not gid.endswith(f"{group}-")

tests/common/test_maths.py

@@ -0,0 +1,65 @@
+from typing import Optional
+
+import pytest
+
+from dodal.common import in_micros, step_to_num
+
+
+@pytest.mark.parametrize(
+    "s,us",
+    [
+        (4.000_001, 4_000_001),
+        (4.999_999, 4_999_999),
+        (4, 4_000_000),
+        (4.000_000_1, 4_000_001),
+        (4.999_999_9, 5_000_000),
+        (0.1, 100_000),
+        (0.000_000_1, 1),
+        (0, 0),
+    ],
+)
+def test_in_micros(s: float, us: int):
+    assert in_micros(s) == us
+
+
+@pytest.mark.parametrize(
+    "s", [-4.000_001, -4.999_999, -4, -4.000_000_5, -4.999_999_9, -4.05]
+)
+def test_in_micros_negative(s: float):
+    with pytest.raises(ValueError):
+        in_micros(s)
+
+
+@pytest.mark.parametrize(
+    "start,stop,step,expected_num,truncated_stop",
+    [
+        (0, 0, 1, 1, None),  # start=stop, 1 point at start
+        (0, 0.5, 1, 1, 0),  # step>length, 1 point at start
+        (0, 1, 1, 2, None),  # stop=start+step, point at start & stop
+        (0, 0.99, 1, 2, 1),  # stop >= start + 0.99*step, included
+        (0, 0.98, 1, 1, 0),  # stop < start + 0.99*step, not included
+        (0, 1.01, 1, 2, 1),  # stop >= start + 0.99*step, included
+        (0, 1.75, 0.25, 8, 1.75),
+        (0, 0, -1, 1, None),  # start=stop, 1 point at start
+        (0, 0.5, -1, 1, 0),  # abs(step)>length, 1 point at start
+        (0, -1, 1, 2, None),  # stop=start+-abs(step), point at start & stop
+        (0, -0.99, 1, 2, -1),  # stop >= start + 0.99*-abs(step), included
+        (0, -0.98, 1, 1, 0),  # stop < start + 0.99*-abs(step), not included
+        (0, -1.01, 1, 2, -1),  # stop >= start + 0.99*-abs(step), included
+        (0, -1.75, 0.25, 8, -1.75),
+        (1, 10, -0.901, 10, 9.109),  # length overrules step for direction
+        (10, 1, -0.901, 10, 1.891),
+    ],
+)
+def test_step_to_num(
+    start: float,
+    stop: float,
+    step: float,
+    expected_num: int,
+    truncated_stop: Optional[float],
+):
+    truncated_stop = stop if truncated_stop is None else truncated_stop
+    actual_start, actual_stop, num = step_to_num(start, stop, step)
+    assert actual_start == start
+    assert actual_stop == truncated_stop
+    assert num == expected_num