Independent magnet axis - #2153
Conversation
…/dodal into add_i06_magnets
…hecked at each step
…/dodal into add_i06_magnets
…ove redundant tests
…reamline movement logic and enhance field limit handling
…nate interface and movement logic
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## add_i06_magnets #2153 +/- ##
================================================
Coverage 99.18% 99.18%
================================================
Files 359 358 -1
Lines 14219 14243 +24
================================================
+ Hits 14103 14127 +24
Misses 116 116 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
oliwenmandiamond
left a comment
oliwenmandiamond
left a comment
There was a problem hiding this comment.
Please see my comments.
| limits: MagnetPosition, | ||
| ) -> None: | ||
|
|
||
| mag_pos_after_move = target.resolve_pos(current_readback) | ||
| self._check_epics_hardware_limits(mag_pos_after_move, limits) | ||
| if mag_pos_after_move.field_magnitude > self.LIMIT: |
There was a problem hiding this comment.
This check_within_limits is done for every step that is produced by move_steps.
_check_epics_hardware_limits should be done once and independent of the these limits. The movement strategy is only checking if the total field or other bits that are mandated by the mode that the IOC does not check.
You have also duplicated the entire API calls. If this either done outside of this entirely, it is done once or if it put in a new method to the base MovementStrategy and then have the API call it once rather making every movement strategy implementation check it (which might be forgotton).
| self.readback = epics_signal_r(float, f"{prefix}-MAG-01:{axis}:RBV") | ||
| self.demand = epics_signal_rw(float, f"{prefix}-MAG-01:{axis}:DMD") | ||
| self.limit = epics_signal_r(float, f"{prefix}-SMC-0{axis_number}:LIM:FIELD:NOW") | ||
| with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL): | ||
| self.ramp_rate = epics_signal_rw( | ||
| float, | ||
| read_pv=f"{prefix}-SMC-0{axis_number}:STS:RAMPRATE:TPM", | ||
| write_pv=f"{prefix}-SMC-0{axis_number}:SET:DMD:RAMPRATE:TPM", |
There was a problem hiding this comment.
As mentioned before, this should not be done. You are combining lots of different base PV's to represent one device when this does not reflect the hardware.
Each magnet axis has a power supply associated with it which then defines the limits and ramp rates. This should be it's own device using -SMC-0X: and then use pass it in as reference to the axis of PV -MAG-01:
| @AsyncStatus.wrap | ||
| async def prepare(self, value: FlyVectorMagnetInfo) -> None: | ||
| """Prepare the magnet for a fly scan. | ||
| The requested target position is validated against the limits of the current | ||
| magnet operating mode. The corresponding movement strategy then generates a | ||
| sequence of intermediate movement steps. | ||
| Each movement step is validated against the latest magnet readback before it | ||
| is applied. The readback position is updated after each step, ensuring that | ||
| subsequent steps are validated against the magnet's actual current position. | ||
| """ | ||
| self._fly_info = value | ||
| current_readback, mode, field_limit, ramp_rate_limit = await asyncio.gather( | ||
| self.get_readback_position(), | ||
| self.mode.get_value(), | ||
| self.get_field_limit(), | ||
| self.get_ramp_rate_limit(), | ||
| ) | ||
| if mode in { | ||
| MagnetMode.UNIAXIAL_X, | ||
| MagnetMode.UNIAXIAL_Y, | ||
| MagnetMode.UNIAXIAL_Z, | ||
| MagnetMode.CUBIC, | ||
| }: | ||
| movement_strategy = self._MODE_MOVEMENT_STRATEGY.get(mode) | ||
| max_ramp_rate: float = getattr(ramp_rate_limit, value.fly_axis) | ||
| if movement_strategy is None: | ||
| raise ValueError( | ||
| f"No movement strategy has been configured for device {self.name} for mode {mode}." | ||
| ) | ||
| if max_ramp_rate < value.ramp_rate or value.ramp_rate <= 0: | ||
| raise ValueError( | ||
| f"Requested ramp rate {value.ramp_rate} exceeds the ramp rate limit" | ||
| f" of {max_ramp_rate} for axis {value.fly_axis}." | ||
| ) | ||
|
|
||
| start_req = MagnetRequest(**{value.fly_axis: value.start_position}) | ||
| end_req = MagnetRequest(**{value.fly_axis: value.end_position}) | ||
|
|
||
| movement_strategy.check_within_limits( | ||
| current_readback, start_req, field_limit | ||
| ) | ||
|
|
||
| start_readback = dataclasses.replace( | ||
| current_readback, **{value.fly_axis: value.start_position} | ||
| ) | ||
|
|
||
| movement_strategy.check_within_limits(start_readback, end_req, field_limit) | ||
|
|
||
| else: | ||
| raise ValueError( | ||
| f"Cannot prepare fly scan in mode {mode}. Only uniaxial and cubic modes are supported for now." | ||
| ) | ||
| start_position = MagnetRequest(**{value.fly_axis: value.start_position}) | ||
| await self.set_within_boundary(start_position) | ||
| fly_axis = getattr(self, value.fly_axis) | ||
| await fly_axis.ramp_rate.set(value.ramp_rate) | ||
|
|
There was a problem hiding this comment.
I don't agree with this approach.
If i go to fly the magnet controller and I forget to change mode, I might fly scan something completely unexpected.
e.g fly scan scmc, I don't know what I am actually fly scanning when running the plan, I have to have made sure I am in the right mode first which you can very easily not do.
Where as if I fly scan scmc.cart.x, I know for a fact I am always fly scanning x. If i fly scan scmc.cart.z, I am always fly scanning z. It just might throw error if in the wrong mode which is safer. The API is consistent and and never changes where as this one is dynamic and will cause problems down the road.
There was a problem hiding this comment.
Check FlyVectorMagnetInfo, you tell the controller which axis you want to fly, this make it easy to add the other modes and axises.
class FlyVectorMagnetInfo(ConfinedModel):
fly_axis: str = Field(frozen=True)
"""Axis to fly along, one of 'x', 'y' or 'z'. Todo theta, phi or rho."""
start_position: float = Field(frozen=True)
"""Start position of the magnet move. in Tesla"""
end_position: float = Field(frozen=True)
"""End position of the magnet move, in Tesla."""
ramp_rate: float = Field(frozen=True, gt=0)
"""Ramp rate of the magnet move, in Tesla/s."""3 participants
Fixes sm-blusky#352
@oliwenmandiamond Not clean yet, but this shows the general idea. Running the dodal connect tests helped me see why you wanted to break the axis into two devices, but to me they represent the field and its limits, so they belong together.
Changes
self_movingwithasyncio.Lock()and updatedset_within_boundary.set_within_boundaryinjection making it impossible to make a field movement without accessingcontroller._start_rampMovementStrategychecks live limits rather than hardcoded limits.Instructions to reviewer on how to test:
Checks for reviewer
dodal connect ${BEAMLINE}