Move util functions from i22 branch of ophyd_async that may be genera…

…lly useful

pyproject.toml

@@ -15,6 +15,7 @@ description = "Common Diamond specific Bluesky plans and functions"
 dependencies = [
     "blueapi",
     "ophyd",
+    "ophyd_async",
     "scanspec"
 ] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]

src/dls_bluesky_core/core/__init__.py

@@ -1,6 +1,15 @@
-from .types import MsgGenerator, PlanGenerator
+from .coordination import group_uuid
+from .maths import step_to_num, in_micros
+from .scanspecs import get_duration
+from .types import MsgGenerator, PlanGenerator, ScanAxis
+
 
 __all__ = [
+    "get_duration",
+    "group_uuid",
+    "in_micros",
     "MsgGenerator",
     "PlanGenerator",
+    "ScanAxis",
+    "step_to_num"
 ]

src/dls_bluesky_core/core/coordination.py

@@ -0,0 +1,14 @@
+import uuid
+
+
+def group_uuid(name: str) -> str:
+    """
+    Returns a unique but human-readable string, to assist debugging orchestrated groups.
+
+    Args:
+        name (str): A human readable name
+
+    Returns:
+        readable_uid (str): name appended with a unique string
+    """
+    return f"{name}-{str(uuid.uuid4())[:6]}"

src/dls_bluesky_core/core/maths.py

@@ -0,0 +1,47 @@
+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 end point (consistent with GDA)
+
+    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, truncated_stop, num = Tuple[float, float, int]
+        start will be returned unchanged
+        truncated_stop = start + num * 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
+    num = int((stop - start) / step + 0.01)
+    return start, start + num * step, num
+
+
+def in_micros(t: float) -> int:
+    """
+    Converts between units of microT and units of T.
+    For example, from microseconds to seconds.
+
+    Args:
+        t (float): A time in microseconds, or other measurement in units of microU
+    Returns:
+        t (int): A time in seconds rounded up to the nearest whole second,
+        or other measurement in units of U, rounded up to the nearest whole U.
+    """
+    return np.ceil(t / 1e6)

src/dls_bluesky_core/core/scanspecs.py

@@ -0,0 +1,32 @@
+from typing import List, Optional
+
+import numpy as np
+from scanspec.core import Frames
+from scanspec.specs import DURATION
+
+
+def get_duration(frames: List[Frames]) -> Optional[float]:
+    """
+    Returns the duration of a number of ScanSpec frames, if known and consistent.
+
+    Args:
+        frames (List[Frames]): A number of Frame objects
+
+    Raises:
+        ValueError: If any frames do not have a Duration defined.
+
+    Returns:
+        duration (float): if all frames have a consistent duration
+        None: otherwise
+
+    """
+    for fs in frames:
+        if DURATION in fs.axes():
+            durations = fs.midpoints[DURATION]
+            first_duration = durations[0]
+            if np.all(durations == first_duration):
+                # Constant duration, return it
+                return first_duration
+            else:
+                return None
+    raise ValueError("Duration not specified in Spec")

src/dls_bluesky_core/core/types.py

@@ -1,8 +1,11 @@
-from typing import Any, Callable, Generator
+from typing import Any, Callable, Generator, Union
 
 from bluesky import Msg
+from ophyd_async.core import Device
+from scanspec.specs import DURATION
 
 #  'A true "plan", usually the output of a generator function'
 MsgGenerator = Generator[Msg, Any, None]
 #  'A function that generates a plan'
 PlanGenerator = Callable[..., MsgGenerator]
+ScanAxis = Union[Device, DURATION]