Modifications to dynamically fetch the encoder start position for the panda for flyscans

src/xpdtools/flyers.py

@@ -125,6 +125,34 @@ def calculate_move_time_for_flyscan(
     return max(fastest_possible_move_time, total_acq_time)
 
 
+def get_zero_encoder_position(
+    current_position: float,
+    start_position: float,
+    encoder_resolution: float,
+    current_encoder_value: int,
+):
+    """Calculate the encoder position corresponding to 0 degrees.
+
+    Parameters
+    ----------
+    current_position : float
+        The current position of the motor.
+    start_position : float
+        The start position of the flyscan.
+    encoder_resolution : float
+        The resolution of the encoder in counts per degree.
+    current_encoder_value : int
+        The current encoder value.
+
+    Returns
+    -------
+    int
+        The encoder position corresponding to 0 degrees.
+    """
+    dist_to_start_in_cts = (current_position - start_position) / encoder_resolution
+    return int(current_encoder_value - dist_to_start_in_cts)
+
+
 def construct_fly_info_models(
     num_pulses: int,
     max_exposure_time: float,

src/xpdtools/motors.py

@@ -82,15 +82,14 @@ class RotationMotor(AsyncEpicsMotor):
     methods that are specific to rotation scans.
     """
 
-    def __init__(self, prefix: str, encoder_pos_at_zero: int = 0, name: str = ""):
+    def __init__(self, prefix: str, name: str = ""):
         super().__init__(prefix, name=name)
         self.encoder_counts_per_rev = derived_signal_r(
             self.get_encoder_counts_per_rev,
             derived_units="counts",
             derived_precision=0,
             encoder_resolution=self.encoder_resolution,
         )
-        self.encoder_pos_at_zero = encoder_pos_at_zero
 
     def get_encoder_counts_per_rev(self, encoder_resolution: float) -> int:
         """Calculate the number of encoder counts per revolution.
@@ -107,22 +106,22 @@ def get_encoder_counts_per_rev(self, encoder_resolution: float) -> int:
         """
         return int(360.0 * encoder_resolution)
 
-    async def get_encoder_value_from_angle(self, angle: float) -> int:
-        """Calculate the encoder value from an angle.
-
-        Parameters
-        ----------
-        angle : float
-            The angle in degrees.
-
-        Returns
-        -------
-        int
-            The encoder value corresponding to the given angle.
-        """
-        encoder_resolution = await self.encoder_resolution.get_value()
-        return get_encoder_value_from_pos(
-            current_position=angle,
-            encoder_resolution=encoder_resolution,
-            encoder_pos_at_zero=self.encoder_pos_at_zero,
-        )
+    # async def get_encoder_value_from_angle(self, angle: float) -> int:
+    #     """Calculate the encoder value from an angle.
+
+    #     Parameters
+    #     ----------
+    #     angle : float
+    #         The angle in degrees.
+
+    #     Returns
+    #     -------
+    #     int
+    #         The encoder value corresponding to the given angle.
+    #     """
+    #     encoder_resolution = await self.encoder_resolution.get_value()
+    #     return get_encoder_value_from_pos(
+    #         current_position=angle,
+    #         encoder_resolution=encoder_resolution,
+    #         encoder_pos_at_zero=self.encoder_pos_at_zero,
+    #     )

src/xpdtools/plans.py

@@ -10,6 +10,7 @@
 from ophyd_async.fastcs.panda import HDFPanda as PandABox
 
 from xpdtools.detectors.utils import get_detector_acq_times
+from xpdtools.flyers import get_zero_encoder_position
 
 from .flyers import (
     SingleAxisFlyscanController,
@@ -60,6 +61,15 @@ def single_axis_flyscan(
 
     # Get the start position in encoder counts
     encoder_res = yield from bps.rd(motor.encoder_resolution)
+    current_motor_pos = yield from bps.rd(motor.user_readback)
+    current_panda_encoder_value = yield from bps.rd(panda.calc[1].out)  # type: ignore
+    panda_encoder_val_at_start = get_zero_encoder_position(
+        current_position=current_motor_pos,
+        start_position=start,
+        encoder_resolution=encoder_res,
+        current_encoder_value=current_panda_encoder_value,
+    )
+
     max_velocity = yield from bps.rd(motor.max_velocity)
 
     acquisition_periods = yield from get_detector_acq_times(detectors)
@@ -82,7 +92,7 @@ def single_axis_flyscan(
         stop_position=stop,
         encoder_resolution=encoder_res,
         max_motor_velocity=max_velocity,
-        encoder_pos_at_zero=motor.encoder_pos_at_zero,
+        encoder_pos_at_zero=panda_encoder_val_at_start,
         acq_time_overhead=acq_time_overhead,
         time_based=time_based,
     )

src/xpdtools/profiles/xpdd.py

@@ -84,7 +84,7 @@
 
 with init_devices(mock=XPDTOOLS_RUNNING_IN_CI):
     panda1 = HDFPanda("XF:28ID2-ES{PANDA:1}:", path_provider)
-    rot_motor = RotationMotor("XF:28IDD-ES:2{Twister}Mtr", encoder_pos_at_zero=211)
+    rot_motor = RotationMotor("XF:28IDD-ES:2{Twister}Mtr")
     step_motor = AsyncEpicsMotor("XF:28IDD-ES:2{Twister}Mtr", name="amazon_motor")
     pilatus1_stats1 = NDStatsIO("XF:28ID2-ES{Pilatus4-Det:1}Stats1:")
     pilatus1 = Pilatus4Detector(
@@ -93,7 +93,7 @@
         plugins={"stats1": pilatus1_stats1},
     )
     pilatus1.add_detector_logics(
-        PluginSignalDataLogic(pilatus1.driver, pilatus1.stats1.total)
+        PluginSignalDataLogic(pilatus1.driver, pilatus1.stats1.total)  # type: ignore
     )
 
 panda_settings = PackagedSettingsProvider()

tests/test_flyers.py

@@ -6,6 +6,7 @@
     SingleAxisFlyscanInfo,
     calculate_move_time_for_flyscan,
     construct_fly_info_models,
+    get_zero_encoder_position,
 )
 
 
@@ -188,3 +189,31 @@ def test_construct_fly_info_models_raises(
             max_motor_velocity,
             encoder_pos_at_zero,
         )
+
+
+@pytest.mark.parametrize(
+    "current_position, start_position, encoder_resolution, current_encoder_value,"
+    "expected_zero_encoder_position",
+    [
+        (10.0, 0.0, 0.1, 100, 0),
+        (5.0, 2.0, 0.2, 50, 35),
+        (20.0, 10.0, 0.5, 200, 180),
+        (15.0, 5.0, 0.1, 150, 50),
+        (8.0, 4.0, 0.2, 80, 60),
+        (183, 0.0, 0.0009, 198353, -4980),
+    ],
+)
+def test_get_zero_encoder_position(
+    current_position: float,
+    start_position: float,
+    encoder_resolution: float,
+    current_encoder_value: int,
+    expected_zero_encoder_position: int,
+):
+    zero_encoder_position = get_zero_encoder_position(
+        current_position=current_position,
+        start_position=start_position,
+        encoder_resolution=encoder_resolution,
+        current_encoder_value=current_encoder_value,
+    )
+    assert zero_encoder_position == expected_zero_encoder_position

tests/test_plans.py

@@ -6,11 +6,14 @@
 import pytest
 from bluesky.run_engine import RunEngine
 from ophyd_async.core import (
+    Device,
+    DeviceVector,
     StaticPathProvider,
     UUIDFilenameProvider,
     callback_on_mock_put,
     init_devices,
     set_mock_value,
+    soft_signal_r_and_setter,
 )
 from ophyd_async.core._mock_signal_utils import (  # noqa: PLC2701
     _get_mock_signal_backend,
@@ -82,9 +85,16 @@ async def devices(tmp_path: Path):
     path_provider = StaticPathProvider(
         UUIDFilenameProvider(), tmp_path, create_dir_depth=-2
     )
+
+    class CalcBlock(Device):
+        def __init__(self):
+            self.out, _ = soft_signal_r_and_setter(int)
+            super().__init__(name="calc1")
+
     async with init_devices(mock=True):
         panda = HDFPanda("PANDA:", path_provider)
-        motor = RotationMotor("MOT:", encoder_pos_at_zero=0, name="rot_motor")
+        panda.calc = DeviceVector({1: CalcBlock()})
+        motor = RotationMotor("MOT:", name="rot_motor")
         pilatus1 = Pilatus4Detector(
             "DET1:",
             ADWriterFactory.hdf(path_provider),