tligui_y/slic
0
0
Fork 0
forked from slic/slic
Code Pull Requests Actions Wiki Activity

started adjustable refactor; moved code to Adjustable for more re-use

Browse Source
This commit is contained in:
augustin_s
2020-07-17 19:01:53 +02:00
parent 289db68b15
commit b2f2ac441b
5 changed files with 86 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
slic/core/adjustable/adjustable.py
+13 -1
View File
@@ -1,7 +1,13 @@
from slic.utils import typename
from .baseadjustable import BaseAdjustable
from .convenience import SpecConvenience
class Adjustable(BaseAdjustable):
class Adjustable(BaseAdjustable, SpecConvenience):
def __init__(self, name=None):
self.name = name
self.current_task = None
def set(self, *args, **kwargs):
return self.set_target_value(*args, **kwargs)
@@ -13,5 +19,11 @@ class Adjustable(BaseAdjustable):
def moving(self):
return self.is_moving()
def __repr__(self):
name = self.name or typename(self)
value = self.get_current_value()
return "{} at {}".format(name, value)
#TODO handle Task creation only here, not in every subclass
slic/core/adjustable/convenience.py
+14 -14
View File
@@ -3,37 +3,37 @@ import colorama
class SpecConvenience:
def mv(self, value):
self._currentChange = self.set_target_value(value)
return self._currentChange
def mv(self, *args, **kwargs):
return self.set_target_value(*args, **kwargs)
def wm(self, *args, **kwargs):
return self.get_current_value(*args, **kwargs)
#TODO
def mvr(self, value, *args, **kwargs):
if (
hasattr(self, "_currentChange")
and self._currentChange
and not (self._currentChange.status() == "done")
):
startvalue = self._currentChange.target
if hasattr(self, "current_task") and self.current_task and not (self.current_task.status() == "done"):
startvalue = self.current_task.target
elif hasattr(self, "is_moving") and not self.is_moving():
startvalue = self.get_current_value(readback=True, *args, **kwargs)
else:
startvalue = self.get_current_value(*args, **kwargs)
self._currentChange = self.set_target_value(value + startvalue, *args, **kwargs)
return self._currentChange
self.current_task = self.set_target_value(value + startvalue, *args, **kwargs)
return self.current_task
def wait(self):
self._currentChange.wait()
if self.current_task:
self.current_task.wait()
def __call__(self, value=None):
if not value is None:
self._currentChange = self.set_target_value(value)
return self._currentChange
return self.set_target_value(value)
else:
return self.get_current_value()
#TODO: if hasattr(Adj, "update_change"):
def umv(self, *args, **kwargs):
self.update_change(*args, **kwargs)
slic/core/adjustable/dummyadjustable.py
+7 -7
View File
@@ -1,24 +1,24 @@
from slic.core.task import Task
from .adjustable import Adjustable
class DummyAdjustable(Adjustable):
def __init__(self, initial_value=0, name="Dummy"):
self.name = name
super().__init__(name)
self._current_value = initial_value
def get_current_value(self):
return self._current_value
def set_target_value(self, value, hold=False):
def changer():
def change():
self._current_value = value
return Task(changer, hold=hold)
self.current_task = task = Task(change, hold=hold)
return task
def __repr__(self):
name = self.name
value = self.get_current_value()
return f"{name} at {value}"
def is_moving(self):
return False
slic/core/adjustable/genericadjustable.py
+20 -15
View File
@@ -1,34 +1,39 @@
from time import sleep
from slic.core.task import Task
from .adjustable import Adjustable
class GenericAdjustable(Adjustable):
def __init__(self, set, get, wait=None):
self.set = set
self.get = get
self.wait = wait if wait is not None else self._default_wait
self.last_target = get()
def set_target_value(self, pos):
self.last_target = pos
return self.set(pos)
def __init__(self, set, get, wait=None, name=None):
super().__init__(name)
self._set = set
self._get = get
self._wait = wait or self._generic_wait
self._last_target = None
def get_current_value(self):
return self.get()
return self._get()
def set_target_value(self, value, hold=False):
self._last_target = value
change = lambda: self._set(value)
self.current_task = task = Task(change, hold=hold)
return task
def is_moving(self):
return not self.wait()
return not self._wait()
def _default_wait(self):
return self.get() == self.last_target
def _generic_wait(self):
if self._last_target is None:
return True
return self._get() == self._last_target
if __name__ == "__main__":
from time import sleep
pos = 0
slic/core/adjustable/pvadjustable.py
+32 -60
View File
@@ -1,80 +1,52 @@
from time import sleep
from types import SimpleNamespace
from epics import PV
from slic.core.task import Task
from .adjustable import Adjustable
class PvAdjustable:
class PvAdjustable(Adjustable):
def __init__(self, pvsetname, pvreadbackname=None, accuracy=None, sleeptime=0, name=None):
self.Id = pvsetname
self.name = name
self.sleeptime = sleeptime
def __init__(self, pvname_setvalue, pvname_readback=None, accuracy=None, name=None):
name = name or pvname_setvalue
super().__init__(name)
self._pv = PV(self.Id)
self._currentChange = None
self.accuracy = accuracy
if pvreadbackname is None:
self._pvreadback = PV(self.Id)
else:
self._pvreadback = PV(pvreadbackname)
pv_setvalue = PV(pvname_setvalue)
pv_readback = PV(pvname_readback) if pvname_readback else pv_setvalue
self.pvnames = SimpleNamespace(
setvalue = pvname_setvalue,
readback = pvname_readback
)
self.pvs = SimpleNamespace(
setvalue = pv_setvalue,
readback = pv_readback
)
def get_current_value(self, readback=True):
if readback:
currval = self._pvreadback.get()
if not readback:
currval = self._pv.get()
return currval
def is_moving(self):
movedone = 1
if self.accuracy is not None:
if (
abs(
self.get_current_value(readback=False)
- self.get_current_value(readback=True)
)
> self.accuracy
):
movedone = 0
return self.pvs.readback.get()
else:
sleep(self.sleeptime)
return not bool(movedone)
def move(self, value):
self._pv.put(value)
sleep(0.1)
while self.is_moving():
sleep(0.1)
return self.pvs.setvalue.get()
def set_target_value(self, value, hold=False):
changer = lambda: self.move(value)
return Task(changer, hold=hold)
def change():
# use_complete=True enables status in PV.put_complete
self.pvs.setvalue.put(value, wait=True, use_complete=True)
self.current_task = task = Task(change, hold=hold)
return task
# spec-inspired convenience methods
def mv(self, value):
self._currentChange = self.set_target_value(value)
def wm(self, *args, **kwargs):
return self.get_current_value(*args, **kwargs)
def mvr(self, value, *args, **kwargs):
if not self.is_moving():
startvalue = self.get_current_value(readback=True, *args, **kwargs)
def is_moving(self):
if self.accuracy is not None:
setvalue = self.get_current_value(readback=False)
readback = self.get_current_value(readback=True)
delta = abs(setvalue - readback)
return delta > self.accuracy
else:
startvalue = self.get_current_value(readback=False, *args, **kwargs)
self._currentChange = self.set_target_value(value + startvalue, *args, **kwargs)
def wait(self):
self._currentChange.wait()
def __repr__(self):
return "%s is at: %s" % (self.Id, self.get_current_value())
return not self.pvs.setvalue.put_complete