Dynamically adjust scale and offset for angle dataset

src/xpdtools/flyers.py

@@ -38,6 +38,10 @@ class SingleAxisFlyscanInfo(ConfinedModel):
     pulse_width: float | int
     pulse_step: float | int
     time_based: bool
+    position_dataset_name: str = "Angle"
+    position_dataset_units: str = "deg"
+    position_scale: float = 1.0
+    position_offset: float = 0.0
 
 
 class SingleAxisFlyscanController(FlyerController[SingleAxisFlyscanInfo]):
@@ -49,9 +53,14 @@ def __init__(self, panda: CommonPandaBlocks) -> None:
     async def prepare(self, value: SingleAxisFlyscanInfo):
         pcomp = self.panda.pcomp[1]
         pulse = self.panda.pulse[1]
+        calc = self.panda.calc[1]  # type: ignore
         coros = [
             pcomp.dir.set(value.direction),
             pcomp.start.set(value.start),
+            calc.dataset.set(value.position_dataset_name),
+            calc.units.set(value.position_dataset_units),
+            calc.scale.set(value.position_scale),
+            calc.offset.set(value.position_offset),
         ]
         if not value.time_based:
             coros.extend(
@@ -163,6 +172,8 @@ def construct_fly_info_models(
     encoder_pos_at_zero: int = 0,
     acq_time_overhead: float = 0.001,
     time_based: bool = False,
+    position_dataset_name: str = "Angle",
+    position_dataset_units: str = "deg",
 ) -> tuple[SingleAxisFlyscanInfo, FlyMotorInfo]:
     """Construct the fly info models for a single axis flyscan.
 
@@ -216,6 +227,10 @@ def construct_fly_info_models(
         pulse_width=pulse_width,
         pulse_step=pulse_step,
         time_based=time_based,
+        position_dataset_name=position_dataset_name,
+        position_dataset_units=position_dataset_units,
+        position_scale=encoder_resolution,
+        position_offset=(-1 * encoder_pos_at_zero * encoder_resolution),
     )
 
     motor_info = FlyMotorInfo(

src/xpdtools/motors.py

@@ -10,6 +10,7 @@
     set_mock_put_proceeds,
     set_mock_value,
 )
+from ophyd_async.epics.core import epics_signal_r
 from ophyd_async.epics.motor import Motor as AsyncEpicsMotor
 
 
@@ -90,6 +91,7 @@ def __init__(self, prefix: str, name: str = ""):
             derived_precision=0,
             encoder_resolution=self.encoder_resolution,
         )
+        self.encoder_counts = epics_signal_r(int, prefix + ".REP")
 
     def get_encoder_counts_per_rev(self, encoder_resolution: float) -> int:
         """Calculate the number of encoder counts per revolution.

tests/test_flyers.py

@@ -61,6 +61,8 @@ def test_calculate_move_time_for_flyscan(
                 pulse_width=1,
                 pulse_step=100,
                 time_based=False,
+                position_scale=0.1,
+                position_offset=0.0,
             ),
             FlyMotorInfo(start_position=0.0, end_position=100.0, time_for_move=2.0),
         ),
@@ -80,6 +82,8 @@ def test_calculate_move_time_for_flyscan(
                 pulse_width=1,
                 pulse_step=225,
                 time_based=False,
+                position_scale=0.1,
+                position_offset=0.0,
             ),
             FlyMotorInfo(start_position=90.0, end_position=0.0, time_for_move=3.0),
         ),
@@ -99,6 +103,8 @@ def test_calculate_move_time_for_flyscan(
                 pulse_width=0.101,
                 pulse_step=0.2,
                 time_based=True,
+                position_scale=10.0,
+                position_offset=0.0,
             ),
             FlyMotorInfo(start_position=0.0, end_position=50.0, time_for_move=2.0),
         ),
@@ -118,6 +124,8 @@ def test_calculate_move_time_for_flyscan(
                 pulse_width=0.051,
                 pulse_step=0.051,
                 time_based=True,
+                position_scale=360 / 70000,
+                position_offset=-39240 * (360 / 70000),
             ),
             FlyMotorInfo(start_position=0.0, end_position=180.0, time_for_move=91.851),
         ),
@@ -151,6 +159,12 @@ def test_construct_fly_info_models(
     assert flyer_info.pulse_width == pytest.approx(expected_flyer_info.pulse_width)
     assert flyer_info.pulse_step == pytest.approx(expected_flyer_info.pulse_step)
     assert flyer_info.time_based == expected_flyer_info.time_based
+    assert flyer_info.position_scale == pytest.approx(
+        expected_flyer_info.position_scale
+    )
+    assert flyer_info.position_offset == pytest.approx(
+        expected_flyer_info.position_offset
+    )
     assert motor_info.start_position == pytest.approx(
         expected_motor_info.start_position
     )

tests/test_plans.py

@@ -14,6 +14,7 @@
     init_devices,
     set_mock_value,
     soft_signal_r_and_setter,
+    soft_signal_rw,
 )
 from ophyd_async.core._mock_signal_utils import (  # noqa: PLC2701
     _get_mock_signal_backend,
@@ -89,6 +90,10 @@ async def devices(tmp_path: Path):
     class CalcBlock(Device):
         def __init__(self):
             self.out, _ = soft_signal_r_and_setter(int)
+            self.dataset = soft_signal_rw(str)
+            self.units = soft_signal_rw(str)
+            self.scale = soft_signal_rw(float)
+            self.offset = soft_signal_rw(float)
             super().__init__(name="calc1")
 
     async with init_devices(mock=True):
@@ -120,6 +125,7 @@ def __init__(self):
 def test_single_axis_flyscan(RE: RunEngine, devices, time_based: bool):
     """Test that single_axis_flyscan runs to completion and emits correct documents."""
     pilatus1, panda, motor = devices
+    print(list(panda.children()))
 
     docs: dict[str, list] = {}
 
@@ -155,6 +161,11 @@ def collect_doc(name, doc):
     assert stop_doc["exit_status"] == "success"
     assert stop_doc["run_start"] == start_doc["uid"]
 
+    # --- Verify PandA position calc scale/offset ---
+    # encoder_resolution=0.1, encoder_pos_at_zero resolves to 0 -> offset 0.0
+    assert asyncio.run(panda.calc[1].scale.get_value()) == pytest.approx(0.1)
+    assert asyncio.run(panda.calc[1].offset.get_value()) == pytest.approx(0.0)
+
     # --- Verify descriptor ---
     assert "descriptor" in docs
     descriptors = docs["descriptor"]