moved Pv adjustable to adjustable
This commit is contained in:
@@ -2,6 +2,7 @@ import subprocess
|
||||
from threading import Thread
|
||||
from epics import PV
|
||||
from .utilities import Changer
|
||||
from ..aliases import Alias
|
||||
|
||||
|
||||
def _keywordChecker(kw_key_list_tups):
|
||||
@@ -9,139 +10,89 @@ def _keywordChecker(kw_key_list_tups):
|
||||
assert tkey in tlist, "Keyword %s should be one of %s" % (tkw, tlist)
|
||||
|
||||
|
||||
class ValueRdback:
|
||||
def __init__(self, pv_value, pv_readback, name=None, elog=None):
|
||||
self.Id = pv_value
|
||||
self._PV_value = PV(pv_value)
|
||||
self._PV_readback = PV(pv_readback)
|
||||
self._elog = elog
|
||||
class PvRecord:
|
||||
def __init__(
|
||||
self,
|
||||
pvsetname,
|
||||
pvreadbackname = None,
|
||||
accuracy = None,
|
||||
name=None,
|
||||
elog=None,
|
||||
):
|
||||
|
||||
# alias_fields={"setpv": pvsetname, "readback": pvreadbackname},
|
||||
# ):
|
||||
self.Id = pvsetname
|
||||
self.name = name
|
||||
self.alias = Alias(name)
|
||||
# for an, af in alias_fields.items():
|
||||
# self.alias.append(
|
||||
# Alias(an, channel=".".join([pvname, af]), channeltype="CA")
|
||||
# )
|
||||
|
||||
self._pv = PV(self.Id)
|
||||
self._currentChange = None
|
||||
self.accuracy = accuracy
|
||||
|
||||
# Conventional methods and properties for all Adjustable objects
|
||||
def changeTo(self, value, hold=False, check=True):
|
||||
""" Adjustable convention"""
|
||||
|
||||
def changer(value):
|
||||
self._status = self._motor.move(value, ignore_limits=(not check), wait=True)
|
||||
self._status_message = _status_messages[self._status]
|
||||
if not self._status == 0:
|
||||
print(self._status_message)
|
||||
|
||||
# changer = lambda value: self._motor.move(\
|
||||
# value, ignore_limits=(not check),
|
||||
# wait=True)
|
||||
return Changer(
|
||||
target=value,
|
||||
parent=self,
|
||||
changer=changer,
|
||||
hold=hold,
|
||||
stopper=self._motor.stop,
|
||||
)
|
||||
|
||||
def stop(self):
|
||||
""" Adjustable convention"""
|
||||
try:
|
||||
self._currentChange.stop()
|
||||
except:
|
||||
self._motor.stop()
|
||||
pass
|
||||
|
||||
def get_current_value(self):
|
||||
""" Adjustable convention"""
|
||||
return self._PV_readback.get()
|
||||
|
||||
def set_current_value(self, value):
|
||||
""" Adjustable convention"""
|
||||
print("not implemented: depends on a defined offset")
|
||||
|
||||
def get_precision(self):
|
||||
""" Adjustable convention"""
|
||||
if isinstance(self._precision, PV):
|
||||
return self._precision.get()
|
||||
if pvreadbackname is None:
|
||||
self._pvreadback = PV(self.Id)
|
||||
else:
|
||||
return self._precision
|
||||
self._pvreadback = PV(pvreadbackname)
|
||||
|
||||
def set_precision(self, value):
|
||||
""" Adjustable convention"""
|
||||
if isinstance(self._precision, PV):
|
||||
self._precision.put(value)
|
||||
else:
|
||||
self._precision = value
|
||||
|
||||
precision = property(get_precision, set_precision)
|
||||
|
||||
def set_speed(self):
|
||||
""" Adjustable convention"""
|
||||
pass
|
||||
|
||||
def get_speed(self):
|
||||
""" Adjustable convention"""
|
||||
pass
|
||||
|
||||
def set_speedMax(self):
|
||||
""" Adjustable convention"""
|
||||
pass
|
||||
def get_current_value(self, readback=True,):
|
||||
if readback:
|
||||
currval = self._pvreadback.get()
|
||||
if not readback:
|
||||
currval = self._pv.get()
|
||||
return currval
|
||||
|
||||
|
||||
def get_moveDone(self):
|
||||
""" Adjustable convention"""
|
||||
""" 0: moving 1: move done"""
|
||||
return PV(str(self.Id + ".DMOV")).value
|
||||
movedone = 1
|
||||
if self.accuracy is not None:
|
||||
if( np.abs(self.get_current_value(readback=False)-self.get_current_value(readback=True)) > self.accuracy ):
|
||||
movedone = 0
|
||||
return movedone
|
||||
|
||||
def set_limits(self, values, posType="user", relative_to_present=False):
|
||||
|
||||
def move(self,value):
|
||||
self._pv.put(value)
|
||||
time.sleep(0.1)
|
||||
while( self.get_moveDone() == 0 ):
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
def changeTo(self, value, hold=False):
|
||||
""" Adjustable convention"""
|
||||
_keywordChecker([("posType", posType, _posTypes)])
|
||||
ll_name, hl_name = "LLM", "HLM"
|
||||
if posType is "dial":
|
||||
ll_name, hl_name = "DLLM", "DHLM"
|
||||
if relative_to_present:
|
||||
v = self.get_current_value(posType=posType)
|
||||
values = [v - values[0], v - values[1]]
|
||||
self._motor.put(ll_name, values[0])
|
||||
self._motor.put(hl_name, values[1])
|
||||
|
||||
def get_limits(self, posType="user"):
|
||||
""" Adjustable convention"""
|
||||
_keywordChecker([("posType", posType, _posTypes)])
|
||||
ll_name, hl_name = "LLM", "HLM"
|
||||
if posType is "dial":
|
||||
ll_name, hl_name = "DLLM", "DHLM"
|
||||
return self._motor.get(ll_name), self._motor.get(hl_name)
|
||||
changer = lambda value: self.move(\
|
||||
value)
|
||||
return Changer(
|
||||
target=value,
|
||||
parent=self,
|
||||
changer=changer,
|
||||
hold=hold,
|
||||
stopper=None)
|
||||
|
||||
# return string with motor value as variable representation
|
||||
def __str__(self):
|
||||
return "Motor is at %s" % self.wm()
|
||||
|
||||
# spec-inspired convenience methods
|
||||
def mv(self,value):
|
||||
self._currentChange = self.changeTo(value)
|
||||
def wm(self,*args,**kwargs):
|
||||
return self.get_current_value(*args,**kwargs)
|
||||
def mvr(self,value,*args,**kwargs):
|
||||
|
||||
if(self.get_moveDone == 1):
|
||||
startvalue = self.get_current_value(readback=True,*args,**kwargs)
|
||||
else:
|
||||
startvalue = self.get_current_value(readback=False,*args,**kwargs)
|
||||
self._currentChange = self.changeTo(value+startvalue,*args,**kwargs)
|
||||
def wait(self):
|
||||
self._currentChange.wait()
|
||||
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
return "%s is at: %s"%(self.Id,self.get_current_value())
|
||||
|
||||
def __call__(self, value):
|
||||
self._currentChange = self.changeTo(value)
|
||||
|
||||
|
||||
class ChangerOld:
|
||||
def __init__(self, target=None, parent=None, mover=None, hold=True, stopper=None):
|
||||
self.target = target
|
||||
self._mover = mover
|
||||
self._stopper = stopper
|
||||
self._thread = Thread(target=self._mover, args=(target,))
|
||||
if not hold:
|
||||
self._thread.start()
|
||||
|
||||
def wait(self):
|
||||
self._thread.join()
|
||||
|
||||
def start(self):
|
||||
self._thread.start()
|
||||
|
||||
def status(self):
|
||||
if self._thread.ident is None:
|
||||
return "waiting"
|
||||
else:
|
||||
if self._isAlive:
|
||||
return "changing"
|
||||
else:
|
||||
return "done"
|
||||
|
||||
def stop(self):
|
||||
self._stopper()
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from ..devices_general.motors import MotorRecord
|
||||
from ..devices_general.detectors import CameraCA, CameraBS
|
||||
from ..devices_general.pv_adjustable import PvRecord
|
||||
from ..devices_general.adjustable import PvRecord
|
||||
from ..aliases import Alias
|
||||
|
||||
# from ..devices_general.epics_wrappers import EnumSelector
|
||||
|
||||
Reference in New Issue
Block a user