DAQ: rewrite of my_motor class to wrap furtehr functionality of the epics class

This commit is contained in:
2026-01-15 17:09:55 +01:00
parent 8a25ae2568
commit fa8d63ba03
+80 -33
View File
@@ -4,50 +4,97 @@ from epics import Motor
class MyMotor(Motor):
def __init__(self, name):
super().__init__(name.upper())
def __init__(self, name, timeout=5.0):
super().__init__(name.upper(), timeout=timeout)
@property
def speed(self):
"""Gets the current motor.slew_speed value"""
return self.slew_speed
@speed.setter
def speed(self, speed):
self.slew_speed = speed
@property
def pvname(self):
return self._prefix[:-1]
def __readback(self):
return self.readback
position = property(__readback)
def __value(self):
return self.drive
value = property(__value)
class BogusMotor:
def __init__(self, name, **kwargs):
self.name = name
self.readback = kwargs.get("readback", 0.0)
self.target = kwargs.get("target", 0.0)
self.velo = kwargs.get("velo", 10.0)
def speed(self, v):
"""Sets the motor slew speed"""
self.slew_speed = v
@property
def position(self):
"""Gets the current motor readback value"""
return self.readback
def move(self, target, wait=False):
@property
def value(self):
"""Gets the current motor.drive value"""
return self.drive
@value.setter
def value(self, v):
"""Sets motor.drive to a set value"""
self.drive = v
def stop(self):
"""Stops the motor"""
self.stop_motor()
@property
def moving(self):
"""Returns True if the motor is moving"""
return bool(self.moving_flag)
@property
def units(self):
"""Returns the units of the motor as a string"""
return self.get("EGU", as_string=True)
@property
def limits(self):
"""Returns (low_limit, high_limit)"""
return self.get('HLM'), self.get('LLM')
@limits.setter
def limits(self, limits):
"""Sets (low_limit, high_limit)"""
low, high = limits
self.put('LLM', low)
self.put('HLM', high)
def move_motor(self, val, relative=False, wait=False, timeout=300.0):
"""
Moves the motor to an absolute or relative position.
:param val: Position to move to
:param relative: If True, moves relative to current position
:param wait: If True, waits for completion (synchronous)
:param timeout: Maximum time to wait for completion
"""
return self.move(val, relative=relative, wait=wait, timeout=timeout)
def home(self, direction='forward', wait=False):
"""
Homes the motor.
:param direction: 'forward' or 'reverse'
"""
field = 'HOMF' if direction == 'forward' else 'HOMR'
self.put(field, 1)
if wait:
time.sleep(abs(target - self.readback) / self.velo)
self.readback = target
self.wait_for_stop()
def refresh(self):
pass
def wait_for_stop(self, timeout=300.0, poll_rate=0.01):
"""
Synchronous wait until the motor stops moving.
"""
start_time = time.time()
while self.moving:
time.sleep(poll_rate)
if time.time() - start_time > timeout:
raise RuntimeError(f"Timeout waiting for motor {self.name} to stop")
def get_position(self, *args, **kwargs):
return self.readback
async def wait_for_stop_async(self, timeout=300.0, poll_rate=0.01):
"""
Asynchronous wait until the motor stops moving.
"""
import asyncio
start_time = time.time()
while self.moving:
await asyncio.sleep(poll_rate)
if time.time() - start_time > timeout:
raise RuntimeError(f"Timeout waiting for motor {self.name} to stop")