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

frappy_psi/electromagnet: improve write safety

Browse Source 
- move mf._data to io_mf.data
- add more info to data: data_size, name_pos
- check if name matches before write

Change-Id: Ibf544acb5f350a5a4dd6f5866148b95bea50ff86
This commit is contained in:
zolliker
2026-06-09 14:21:22 +02:00
parent 1b17360f0b
commit 2680f8dafb
1 changed files with 59 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frappy_psi/electromagnet.py
+59 -50
View File
@@ -20,7 +20,7 @@
import struct
from frappy.core import BytesIO, HasIO, Drivable, Parameter, Property, IntRange, FloatRange, BoolType, \
Attached, IDLE, BUSY, ERROR
Attached, IDLE, BUSY, ERROR, Command, TupleOf, StringType
from frappy.errors import CommunicationFailedError
@@ -37,34 +37,55 @@ 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)
def checkHWIdent(self):
reply = self.send_fetch_message(1) # length in words
pattern = struct.pack('>2s14B', b'S5', 16, 1, 3, 6, 15, 3, 0, 255, 7, 0, 0, 0, 0, 0)
if reply[0:16] != pattern:
raise CommunicationFailedError(f'bad response: {reply}')
self._db_length = struct.unpack('>H', reply[16:18])[0] // 2 # length in words (1 word = 2 bytes = 16 bit)
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()
def send_fetch_message(self, length):
# length in words
msg = struct.pack('>2s10BH2B', b'S5', 16, 1, 3, 5, 3, 8, 1, self.db, 0, 0, length, 255, 2)
def send_fetch_message(self, start, length):
"""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)
return reply
if reply[:16] != struct.pack('>2s8B2H2B', b'S5', 16, 1, 3, 6, 15, 3, 0, 255, 7, 0, 0, 0, 0, 0):
raise CommunicationFailedError('bad reply header: %r' % reply[:16])
return reply[16:]
def send_write_message(self, start, typ, value):
if typ == 4: # bool
length = 2
@Command(argument=TupleOf(IntRange(0,999), IntRange(0,255), IntRange(0, 255)),
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)
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])
def send_write_message(self, key, value):
info = self.data[key]
name = info['name'].encode('latin-1')
length = len(name)
# check if name matches
reply = self.send_fetch_message(info['name_pos'], length // 2 + 2)
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
content = struct.pack('>H', bool(value) * 256)
elif typ == 2: # int
length = 4
content = struct.pack('>I', value)
elif typ == 3: # float
length = 4
start += 2
content = struct.pack('>f', value)
else:
raise ValueError('unknown type')
msg = struct.pack('>2s8BHH2B', b'S5', 16, 1, 3, 3, 3, 8, 1, self.db, start, length, 255, 2)
msg = struct.pack('>2s8BHH2B', b'S5', 16, 1, 3, 3, 3, 8, 1, self.db,
info['data_start'], info['data_size'], 255, 2)
self.communicate(msg + content, 16)
def get_string(self, pos, reply):
@@ -75,36 +96,33 @@ class IO(BytesIO):
return pos, value
def get_data(self):
reply = self.send_fetch_message(self._db_length)
reply = self.send_fetch_message(0, self._db_length)
# self.log.info('get_data reply %s', reply)
pos = 16 + 2
pos = 2
result = {}
while pos < len(reply):
typ, alarm = reply[pos:pos+2]
pos += 2
name_pos = pos
pos, name = self.get_string(pos, reply)
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-16+2)
if typ == 1:
# value = struct.unpack('>H', reply[pos:pos+2])[0]
value = None
pos += 2
elif typ == 4:
self.log.debug('pos %d', pos)
data_pos = pos
if typ in (1, 4):
length = 2
value = bool(struct.unpack('>H', reply[pos:pos+length])[0])
pos += length
elif typ == 2:
length = 4
value = struct.unpack('>I', reply[pos:pos+length])[0]
pos += length
elif typ == 3:
length = 6
_, value = struct.unpack('>Hf', reply[pos:pos+length])
pos += length
data_pos = pos + 2 # skip 2 bytes with unknown purpose
length = 4
value = struct.unpack('>Hf', reply[pos:pos+length])[0]
else:
raise ValueError('unknown type')
pos = data_pos + length
res = {
'name': name,
'unit': unit,
@@ -113,11 +131,13 @@ class IO(BytesIO):
'typ': typ,
'alarm': alarm,
'value': value,
'byte_start': pos - length
'name_pos': name_pos,
'data_pos': data_pos,
'data_size': length // 2,
}
result[name, description] = res
self.log.debug(result)
return result
self.data = result
class Magnet(HasIO, Drivable):
@@ -149,19 +169,9 @@ class Magnet(HasIO, Drivable):
current_timeout = Parameter('magnet current timeout flag', datatype=BoolType())
polarity = Parameter('positive polarity', datatype=BoolType())
_busy = False
_data = None
def initialReads(self):
self._data = self.io.get_data()
return super().initialReads()
def _write_data(self, key, value):
start = self._data[key]['byte_start']
typ = self._data[key]['typ']
self.io.send_write_message(start, typ, value)
def extract_data(self, key):
val = self._data[key]['value']
val = self.io.data[key]['value']
return float(val)
def read_value(self):
@@ -175,13 +185,13 @@ class Magnet(HasIO, Drivable):
self._busy = True
self.setFastPoll(True)
self.write_main_switch(True)
self._write_data(('setpoint', 'Magnetcurrent'), target)
self.io.send_write_message(('setpoint', 'Magnetcurrent'), target)
def read_ramp_time(self):
return self.extract_data(('time', 'Current_ramp@75A'))
def write_ramp_time(self, ramp_time):
self._write_data(('time', 'Current_ramp@75A'), ramp_time)
self.io.send_write_message(('time', 'Current_ramp@75A'), ramp_time)
def read_collective_fault(self):
return self.extract_data(('alert', 'collective_fault'))
@@ -190,7 +200,7 @@ class Magnet(HasIO, Drivable):
return self.extract_data(('alert', 'emergency_stop'))
def write_main_switch(self, val):
self._write_data(('setpoint', 'Magnetcurrent'), val)
self.io.send_write_message(('setpoint', 'Magnetcurrent'), val)
def read_main_switch(self):
return self.extract_data(('main_switch', 'on'))
@@ -228,12 +238,11 @@ class Magnet(HasIO, Drivable):
def stop(self):
"""stop ramp"""
val = self.read_value()
self._write_data(('setpoint', 'Magnetcurrent'), val)
self.target = val
self.target = self.read_value()
self.io.send_write_message(('setpoint', 'Magnetcurrent'), self.target)
def doPoll(self):
self._data = self.io.get_data()
self.io.get_data()
super().doPoll()