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

frappy_psi.electromagnet: further fixes

Browse Source
This commit is contained in:
zolliker
2026-05-15 17:36:37 +02:00
parent d884c00204
commit 7a5d71d1cf
3 changed files with 40 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cfg/emagnet_cfg.py
+2 -2
View File
@@ -1,6 +1,6 @@
Node('electromagnet.psi.ch',
'electromagnet SPS',
'tcp://5001',
'tcp://5000',
)
IO('io_mf', uri='192.168.2.2:2000')
@@ -27,4 +27,4 @@ Mod('sample',
'frappy_psi.electromagnet.SampleHolder',
'sample holder position',
motor='mot',
)
)
cfg/phytron_cfg.py
+5 -8
View File
@@ -3,16 +3,13 @@ Node('phytron_test.psi.ch',
interface='tcp://5000',
)
Mod('drv_io',
'frappy_psi.phytron.PhytronIO',
'',
uri='ma7-ts.psi.ch:3007',
)
IO('io_drv', 'serial:///dev/ttyS2?baudrate=115200')
Mod('drv',
'frappy_psi.phytron.Motor',
'a phytron motor',
io='drv_io',
abslimits=(-180.0, 360.0),
encoder_mode='CHECK',
io='io_drv',
# abslimits=(-180.0, 360.0),
encoder_mode='READ',
check_limit_switches = True,
)
frappy_psi/electromagnet.py
+33 -22
View File
@@ -38,8 +38,10 @@ class IO(BytesIO):
db = Property('database number', datatype=IntRange(0, 255), default=200)
_db_length = 0
fetch_header = struct.pack('>2s14B', b'S5', 16, 1, 3, 6, 15, 3, 0, 255, 7, 0, 0, 0, 0, 0)
data = None
def checkHWIdent(self):
self.data = {}
reply = self.send_fetch_message(0, 1) # length in words
self._db_length = struct.unpack('>H', reply)[0] // 2 # length in words (1 word = 2 bytes = 16 bit)
self.get_data()
@@ -48,7 +50,7 @@ class IO(BytesIO):
"""fetch data from start (byte offset) with length (word length = byte-length / 2)"""
msg = struct.pack('>2s8B2H2B', b'S5', 16, 1, 3, 5, 3, 8, 1, self.db, start, length, 255, 2)
reply = self.communicate(msg, 16 + length * 2)
if reply[:16] != struct.pack('>2s8B2H2B', b'S5', 16, 1, 3, 6, 15, 3, 0, 255, 7, 0, 0, 0, 0, 0):
if reply[:16] != self.fetch_header:
raise CommunicationFailedError('bad reply header: %r' % reply[:16])
return reply[16:]
@@ -56,16 +58,15 @@ class IO(BytesIO):
result=StringType())
def write(self, start, byte1, byte2):
"""write 2 bytes, for debug purposes"""
msg = struct.pack('>2s8BHH4B', b'S5', 16, 1, 3, 3, 3, 8, 1, self.db, start-16, 1, 255, 2, byte1, byte2)
msg = struct.pack('>2s8BHH4B', b'S5', 16, 1, 3, 3, 3, 8, 1, self.db, start, 1, 255, 2, byte1, byte2)
return repr(self.communicate(msg, 16))
@Command(argument=IntRange(0, 999), result=StringType()) # result = TupleOf(*[IntRange(0, 255) for _ in range(4)])
def read(self, start):
""""read 4 bytes, for debug purposes"""
msg = struct.pack('>2s8BHH2B', b'S5', 16, 1, 3, 5, 3, 8, 1, self.db, start-16, 2, 255, 2)
reply = self.communicate(msg, 20)
self.log.warn('%r', reply)
return '%2.2x %2.2x %2.2x %2.2x' % tuple(reply[16:20])
""""read 16 bytes, for debug purposes"""
msg = struct.pack('>2s8BHH2B', b'S5', 16, 1, 3, 5, 3, 8, 1, self.db, start, 8, 255, 2)
reply = self.communicate(msg, 16+16)
return ' '.join(f'{v:02x}' for v in reply[16:])
def send_write_message(self, key, value):
info = self.data[key]
@@ -76,17 +77,24 @@ class IO(BytesIO):
if reply[1] != length and reply[2:2+length] != name:
raise CommunicationFailedError('communication error, name mismatch')
typ = info['typ']
if typ in (1, 4): # bool, typ 1: unknown
if typ == 1: # int16 (or uint16 ?)
content = struct.pack('>h', value)
if typ == 4: # bool
content = struct.pack('>H', bool(value) * 256)
elif typ == 2: # int
content = struct.pack('>I', value)
elif typ == 2: # int32 (or uint32 ?)
content = struct.pack('>i', value)
elif typ == 3: # float
content = struct.pack('>f', value)
else:
raise ValueError('unknown type')
self.log.debug('write %s: %r %r', key, info['data_pos'], content)
msg = struct.pack('>2s8BHH2B', b'S5', 16, 1, 3, 3, 3, 8, 1, self.db,
info['data_start'], info['data_size'], 255, 2)
info['data_pos'], info['data_size'], 255, 2)
reply = self.send_fetch_message(info['data_pos']-16, 16)
self.log.debug('old %r %r', reply[:16], reply[16:])
self.communicate(msg + content, 16)
reply = self.send_fetch_message(info['data_pos']-16, 16)
self.log.debug('new %r %r', reply[:16], reply[16:])
def get_string(self, pos, reply):
size, length = reply[pos:pos+2] # reserved space, actual length (containing info)
@@ -108,18 +116,20 @@ class IO(BytesIO):
pos, unit = self.get_string(pos, reply)
pos, description = self.get_string(pos, reply)
pos, reference = self.get_string(pos, reply)
self.log.debug('pos %d', pos)
data_pos = pos
if typ in (1, 4):
if typ == 1: # int16 (or uint16?)
length = 2
value = struct.unpack('>h', reply[pos:pos+length])[0]
elif typ == 4: # bool
length = 2
value = bool(struct.unpack('>H', reply[pos:pos+length])[0])
elif typ == 2:
elif typ == 2: # int32 (or uint32?)
length = 4
value = struct.unpack('>I', reply[pos:pos+length])[0]
elif typ == 3:
value = struct.unpack('>i', reply[pos:pos+length])[0]
elif typ == 3: # float32
data_pos = pos + 2 # skip 2 bytes with unknown purpose
length = 4
value = struct.unpack('>Hf', reply[pos:pos+length])[0]
value = struct.unpack('>f', reply[data_pos:data_pos+length])[0]
else:
raise ValueError('unknown type')
pos = data_pos + length
@@ -135,8 +145,9 @@ class IO(BytesIO):
'data_pos': data_pos,
'data_size': length // 2,
}
if res != self.data.get((name, description)):
self.log.debug('%r, %r', reply[data_pos:pos], res)
result[name, description] = res
self.log.debug(result)
self.data = result
@@ -158,7 +169,7 @@ class Magnet(HasIO, Drivable):
target = Parameter('target', datatype=FloatRange(unit='T'))
value = Parameter('value', datatype=FloatRange(unit='T'))
scale = Parameter('scale for actual value', datatype=FloatRange(unit='T/A'), default=1) # TODO: scale factor
scale = Parameter('scale for actual value', datatype=FloatRange(unit='T/A'), default=0.3/75)
ramp_time = Parameter('time to ramp to 75A', datatype=FloatRange(unit='sec'), readonly=False)
main_switch = Parameter('main switch', datatype=BoolType(), readonly=False)
emergency_stop = Parameter('state of emergency stop button', datatype=BoolType())
@@ -179,13 +190,13 @@ class Magnet(HasIO, Drivable):
return current * self.scale
def read_target(self):
return self.extract_data(('setpoint', 'Magnetcurrent'))
return self.extract_data(('setpoint', 'Magnetcurrent')) * self.scale
def write_target(self, target):
self._busy = True
self.setFastPoll(True)
self.write_main_switch(True)
self.io.send_write_message(('setpoint', 'Magnetcurrent'), target)
self.io.send_write_message(('setpoint', 'Magnetcurrent'), target / self.scale)
def read_ramp_time(self):
return self.extract_data(('time', 'Current_ramp@75A'))
@@ -200,7 +211,7 @@ class Magnet(HasIO, Drivable):
return self.extract_data(('alert', 'emergency_stop'))
def write_main_switch(self, val):
self.io.send_write_message(('setpoint', 'Magnetcurrent'), val)
self.io.send_write_message(('main_switch', 'on'), val)
def read_main_switch(self):
return self.extract_data(('main_switch', 'on'))