132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
class SmartMagnet(DeviceBase):
|
|
def __init__(self, name):
|
|
#DeviceBase.__init__(self, name, get_plugin_manager().getDynamicClass("SmartMagnetConfig")())
|
|
DeviceBase.__init__(self, name, DeviceConfig({
|
|
"holdingCurrent":0.0,
|
|
"restingCurrent":0.0,
|
|
"mountCurrent":0.0,
|
|
"unmountCurrent":0.0,
|
|
"reverseCurrent":0.0,
|
|
"reverseTime":0.0,
|
|
}))
|
|
|
|
def doInitialize(self):
|
|
super(SmartMagnet, self).doInitialize()
|
|
self.get_current()
|
|
|
|
def set_current(self, current):
|
|
self.setCache(current, None)
|
|
smc_current.write(current)
|
|
|
|
def get_current(self):
|
|
cur = smc_current.read()
|
|
self.setCache(cur, None)
|
|
return cur
|
|
|
|
def get_current_rb(self):
|
|
self.assert_status()
|
|
return smc_current_rb.read()
|
|
|
|
def get_status(self):
|
|
return smc_magnet_status.read()
|
|
|
|
def assert_status(self):
|
|
pass
|
|
#if self.get_status() == False:
|
|
# raise Exception("Smart Magnet is in faulty status.")
|
|
|
|
def is_mounted(self):
|
|
self.assert_status()
|
|
m1 = smc_mounted_1.read()
|
|
m2 = smc_mounted_2.read()
|
|
if m2==m1:
|
|
raise Exception("Smart Magnet has invalid detection.")
|
|
return m2
|
|
|
|
def set_supress(self, value):
|
|
smc_sup_det.write(value)
|
|
|
|
def get_supress(self):
|
|
return smc_sup_det.read()
|
|
|
|
def check_mounted(self, idle_time =1.0, timeout = -1, interval = 0.1):
|
|
self.assert_status()
|
|
start = time.time()
|
|
last = None
|
|
while True:
|
|
try:
|
|
det = self.is_mounted()
|
|
except:
|
|
det = None
|
|
if det != last:
|
|
settling_timestamp = time.time()
|
|
last = det
|
|
else:
|
|
if det is not None:
|
|
if (time.time()-settling_timestamp > idle_time):
|
|
print "got result " + str(det)
|
|
return det
|
|
if timeout >= 0:
|
|
if (time.time() - start) > timeout:
|
|
raise Exception("Timeout waiting for Smart Magnet detection.")
|
|
print "smc trying hard: det=" + str(det) + " last="+str(last)
|
|
time.sleep(interval)
|
|
|
|
|
|
def doUpdate(self):
|
|
try:
|
|
if self.get_supress():
|
|
self.setState(State.Paused)
|
|
elif self.is_mounted():
|
|
self.setState(State.Busy)
|
|
else:
|
|
self.setState(State.Ready)
|
|
except:
|
|
self.setState(State.Fault)
|
|
|
|
def set_holding_current(self):
|
|
self.set_current(self.config.getFieldValue("holdingCurrent"))
|
|
|
|
def set_resting_current(self):
|
|
self.set_current(self.config.getFieldValue("restingCurrent"))
|
|
|
|
def set_mount_current(self):
|
|
self.set_current(self.config.getFieldValue("mountCurrent"))
|
|
|
|
def set_unmount_current(self):
|
|
self.set_current(self.config.getFieldValue("unmountCurrent"))
|
|
|
|
def set_reverse_current(self):
|
|
self.set_current(self.config.getFieldValue("reverseCurrent"))
|
|
|
|
def set_default_current(self):
|
|
if self.is_mounted():
|
|
self.set_holding_current()
|
|
else:
|
|
self.set_resting_current()
|
|
|
|
def is_resting_current(self):
|
|
is_resting = 2.0 > abs( float(self.get_current()) - float(self.config.getFieldValue("restingCurrent")) )
|
|
return is_resting
|
|
|
|
|
|
def apply_reverse(self):
|
|
reverse_wait = float(self.config.getFieldValue("reverseTime"))
|
|
if reverse_wait >0:
|
|
self.set_reverse_current()
|
|
time.sleep(reverse_wait)
|
|
|
|
#Setting resting current to better detect sample
|
|
def apply_resting(self):
|
|
if not self.is_resting_current():
|
|
self.set_resting_current()
|
|
|
|
|
|
|
|
|
|
add_device(SmartMagnet("smart_magnet"), force = True)
|
|
|
|
smart_magnet.polling = 1000
|
|
|
|
smart_magnet.set_default_current()
|