linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix triton

Browse Source 
tested at warm kelvinox triton system

Change-Id: I1c8064c78846db5b077363be3a8ecff5e2766ddb
This commit is contained in:
zolliker 2022-05-03 16:48:33 +02:00
parent f9db90d89b
commit 2bdfcf10a3
2 changed files with 87 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
secop_psi/mercury.py
View File

@ -61,7 +61,7 @@ fast_slow = Mapped(ON=0, OFF=1) # maps OIs slow=ON/fast=OFF to sample_rate.slow
class IO(StringIO): class IO(StringIO):
identification = [('*IDN?', r'IDN:OXFORD INSTRUMENTS:MERCURY*')] identification = [('*IDN?', r'IDN:OXFORD INSTRUMENTS:*')]
class MercuryChannel(HasIO): class MercuryChannel(HasIO):
@ -70,7 +70,6 @@ class MercuryChannel(HasIO):
example: DB6.T1,DB1.H1 example: DB6.T1,DB1.H1
slot ids for sensor (and control output)''', slot ids for sensor (and control output)''',
StringType()) StringType())
channel_name = Parameter('mercury nick name', StringType(), default='')
channel_type = '' #: channel type(s) for sensor (and control) e.g. TEMP,HTR or PRES,AUX channel_type = '' #: channel type(s) for sensor (and control) e.g. TEMP,HTR or PRES,AUX
def _complete_adr(self, adr): def _complete_adr(self, adr):
@ -111,6 +110,7 @@ class MercuryChannel(HasIO):
assert keys == tuple(names) assert keys == tuple(names)
return tuple(convert(r) for r in result) return tuple(convert(r) for r in result)
except (AssertionError, AttributeError, ValueError): except (AssertionError, AttributeError, ValueError):
time.sleep(0.1) # in case this was the answer of a previous command
raise HardwareError('invalid reply %r to cmd %r' % (reply, cmd)) from None raise HardwareError('invalid reply %r to cmd %r' % (reply, cmd)) from None
def multichange(self, adr, values, convert=as_float): def multichange(self, adr, values, convert=as_float):
@ -142,6 +142,7 @@ class MercuryChannel(HasIO):
assert any(v == 'VALID' for v in valid) assert any(v == 'VALID' for v in valid)
return tuple(convert(r) for r in result) return tuple(convert(r) for r in result)
except (AssertionError, AttributeError, ValueError) as e: except (AssertionError, AttributeError, ValueError) as e:
time.sleep(0.1) # in case of missed replies this might help to skip garbage
raise HardwareError('invalid reply %r to cmd %r' % (reply, cmd)) from e raise HardwareError('invalid reply %r to cmd %r' % (reply, cmd)) from e
def query(self, adr, convert=as_float): def query(self, adr, convert=as_float):
@ -156,11 +157,6 @@ class MercuryChannel(HasIO):
adr, _, name = adr.rpartition(':') adr, _, name = adr.rpartition(':')
return self.multichange(adr, [(name, value)], convert)[0] return self.multichange(adr, [(name, value)], convert)[0]
def read_channel_name(self):
if self.channel_name:
return Done # channel name will not change
return self.query('0:NICK', as_string)
class TemperatureSensor(MercuryChannel, Readable): class TemperatureSensor(MercuryChannel, Readable):
channel_type = 'TEMP' channel_type = 'TEMP'

147
secop_psi/triton.py
View File

@ -21,19 +21,34 @@
"""oxford instruments triton (kelvinoxjt dil)""" """oxford instruments triton (kelvinoxjt dil)"""
from secop.core import Drivable, HasIO, Writable, \ from secop.core import Drivable, HasIO, Writable, \
Parameter, Property, Readable, StringIO, Attached, Done, IDLE, nopoll Parameter, Property, Readable, StringIO, Attached, Done, IDLE, WARN, nopoll
from secop.datatypes import EnumType, FloatRange, StringType, StructOf, BoolType from secop.datatypes import EnumType, FloatRange, StringType, StructOf, BoolType
from secop.lib.enum import Enum
from secop_psi.mercury import MercuryChannel, Mapped, off_on from secop_psi.mercury import MercuryChannel, Mapped, off_on
import secop_psi.mercury as mercury import secop_psi.mercury as mercury
actions = Enum(none=0, condense=1, circulate=2, collect=3)
opened_closed = Mapped(opened=False, closed=True) open_close = Mapped(CLOSE=False, OPEN=True)
open_close = Mapped(open=False, clos=True) actions_map = Mapped(NONE=actions.none, COND=actions.condense, COLL=actions.collect)
class DilState(MercuryChannel): class Action(MercuryChannel):
pass channel_type = 'ACTN'
# states: value = Parameter('running action', EnumType(actions))
target = Parameter('valve target', EnumType(none=0, condense=1, collect=3), readonly=False)
_target = 0
def read_value(self):
return self.query('SYS:DR:ACTN', actions_map)
def read_target(self):
return self._target
def write_target(self, value):
self._target = value
return self.change('SYS:DR:ACTN', value, actions_map)
# actions:
# CLDN (cool down) # CLDN (cool down)
# PCL (precool automation) # PCL (precool automation)
# COND (condense mixture) # COND (condense mixture)
@ -43,7 +58,6 @@ class DilState(MercuryChannel):
# COLL (collect mixture) # COLL (collect mixture)
# EPCL (empty pre-coll automation) # EPCL (empty pre-coll automation)
# STOP # STOP
#
class Valve(MercuryChannel, Writable): class Valve(MercuryChannel, Writable):
@ -52,7 +66,7 @@ class Valve(MercuryChannel, Writable):
target = Parameter('valve target', EnumType(close=0, open=1)) target = Parameter('valve target', EnumType(close=0, open=1))
def read_value(self): def read_value(self):
return self.query('VALV:SIG:STATE', opened_closed) return self.query('VALV:SIG:STATE', open_close)
def write_target(self, value): def write_target(self, value):
return self.change('VALV:SIG:STATE', value, open_close) return self.change('VALV:SIG:STATE', value, open_close)
@ -60,15 +74,8 @@ class Valve(MercuryChannel, Writable):
class Pump(MercuryChannel, Writable): class Pump(MercuryChannel, Writable):
channel_type = 'PUMP' channel_type = 'PUMP'
value = Parameter('pump state', EnumType(closed=0, opened=1)) value = Parameter('pump state', EnumType(off=0, on=1))
target = Parameter('pump target', EnumType(close=0, open=1)) target = Parameter('pump target', EnumType(off=0, on=1))
power = Parameter('pump power', FloatRange(unit='W'))
freq = Parameter('pump frequency', FloatRange(unit='Hz'))
powerstage_temp = Parameter('temperature of power stage', FloatRange(unit='K'))
motor_temp = Parameter('temperature of motor', FloatRange(unit='K'))
bearing_temp = Parameter('temperature of bearing', FloatRange(unit='K'))
pumpbase_temp = Parameter('temperature of pump base', FloatRange(unit='K'))
electronics_temp = Parameter('temperature of electronics', FloatRange(unit='K'))
def read_value(self): def read_value(self):
return self.query('PUMP:SIG:STATE', off_on) return self.query('PUMP:SIG:STATE', off_on)
@ -77,8 +84,23 @@ class Pump(MercuryChannel, Writable):
return self.change('PUMP:SIG:STATE', value, off_on) return self.change('PUMP:SIG:STATE', value, off_on)
def read_status(self): def read_status(self):
# TODO: check possible status values return IDLE, ''
return self.WARN, self.query('PUMP:SIG:STATUS')
class TurboPump(Pump):
power = Parameter('pump power', FloatRange(unit='W'))
freq = Parameter('pump frequency', FloatRange(unit='Hz'))
powerstage_temp = Parameter('temperature of power stage', FloatRange(unit='K'))
motor_temp = Parameter('temperature of motor', FloatRange(unit='K'))
bearing_temp = Parameter('temperature of bearing', FloatRange(unit='K'))
pumpbase_temp = Parameter('temperature of pump base', FloatRange(unit='K'))
electronics_temp = Parameter('temperature of electronics', FloatRange(unit='K'))
def read_status(self):
status = self.query('PUMP:STATUS', str)
if status == 'OK':
return IDLE, ''
return WARN, status
def read_power(self): def read_power(self):
return self.query('PUMP:SIG:POWR') return self.query('PUMP:SIG:POWR')
@ -102,44 +124,44 @@ class Pump(MercuryChannel, Writable):
return self.query('PUMP:SIG:ET') return self.query('PUMP:SIG:ET')
class PulseTubeCompressor(MercuryChannel, Writable): # class PulseTubeCompressor(MercuryChannel, Writable):
channel_type = 'PTC' # channel_type = 'PTC'
value = Parameter('compressor state', EnumType(closed=0, opened=1)) # value = Parameter('compressor state', EnumType(closed=0, opened=1))
target = Parameter('compressor target', EnumType(close=0, open=1)) # target = Parameter('compressor target', EnumType(close=0, open=1))
water_in_temp = Parameter('temperature of water inlet', FloatRange(unit='K')) # water_in_temp = Parameter('temperature of water inlet', FloatRange(unit='K'))
water_out_temp = Parameter('temperature of water outlet', FloatRange(unit='K')) # water_out_temp = Parameter('temperature of water outlet', FloatRange(unit='K'))
helium_temp = Parameter('temperature of helium', FloatRange(unit='K')) # helium_temp = Parameter('temperature of helium', FloatRange(unit='K'))
helium_low_pressure = Parameter('helium pressure (low side)', FloatRange(unit='mbar')) # helium_low_pressure = Parameter('helium pressure (low side)', FloatRange(unit='mbar'))
helium_high_pressure = Parameter('helium pressure (high side)', FloatRange(unit='mbar')) # helium_high_pressure = Parameter('helium pressure (high side)', FloatRange(unit='mbar'))
motor_current = Parameter('motor current', FloatRange(unit='A')) # motor_current = Parameter('motor current', FloatRange(unit='A'))
#
def read_value(self): # def read_value(self):
return self.query('PTC:SIG:STATE', off_on) # return self.query('PTC:SIG:STATE', off_on)
#
def write_target(self, value): # def write_target(self, value):
return self.change('PTC:SIG:STATE', value, off_on) # return self.change('PTC:SIG:STATE', value, off_on)
#
def read_status(self): # def read_status(self):
# TODO: check possible status values # # TODO: check possible status values
return self.WARN, self.query('PTC:SIG:STATUS') # return self.WARN, self.query('PTC:SIG:STATUS')
#
def read_water_in_temp(self): # def read_water_in_temp(self):
return self.query('PTC:SIG:WIT') # return self.query('PTC:SIG:WIT')
#
def read_water_out_temp(self): # def read_water_out_temp(self):
return self.query('PTC:SIG:WOT') # return self.query('PTC:SIG:WOT')
#
def read_helium_temp(self): # def read_helium_temp(self):
return self.query('PTC:SIG:HT') # return self.query('PTC:SIG:HT')
#
def read_helium_low_pressure(self): # def read_helium_low_pressure(self):
return self.query('PTC:SIG:HLP') # return self.query('PTC:SIG:HLP')
#
def read_helium_high_pressure(self): # def read_helium_high_pressure(self):
return self.query('PTC:SIG:HHP') # return self.query('PTC:SIG:HHP')
#
def read_motor_current(self): # def read_motor_current(self):
return self.query('PTC:SIG:MCUR') # return self.query('PTC:SIG:MCUR')
class FlowMeter(MercuryChannel, Readable): class FlowMeter(MercuryChannel, Readable):
@ -163,18 +185,17 @@ class TemperatureSensor(mercury.TemperatureSensor):
return self.change('TEMP:FILT:TIME', value) return self.change('TEMP:FILT:TIME', value)
def read_dwell_time(self): def read_dwell_time(self):
return self.query('TEMP:FILT:DWEL') return self.query('TEMP:MEAS:DWEL')
def write_dwell_time(self, value): def write_dwell_time(self, value):
self.change('TEMP:FILT:WIN', 80) self.change('TEMP:FILT:WIN', 80)
return self.change('TEMP:FILT:DWEL', value) return self.change('TEMP:MEAS:DWEL', value)
def read_pause_time(self): def read_pause_time(self):
return self.query('TEMP:FILT:PAUS') return self.query('TEMP:MEAS:PAUS')
def write_pause_time(self, value): def write_pause_time(self, value):
self.change('TEMP:FILT:WIN', 80) return self.change('TEMP:MEAS:PAUS', value)
return self.change('TEMP:FILT:PAUS', value)
class TemperatureLoop(mercury.TemperatureLoop): class TemperatureLoop(mercury.TemperatureLoop):