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

frappy_psi.pfeiffer_new: calculate crc in PfeifferProtocol

Browse Source
This commit is contained in:
zolliker 2025-05-22 15:08:41 +02:00
parent 6fed5a3651
commit 14f61b01db
1 changed files with 45 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
frappy_psi/pfeiffer_new.py
View File

@ -12,25 +12,29 @@ from frappy.errors import CommunicationFailedError
class PfeifferProtocol(StringIO):
end_of_line = '\r'
@staticmethod
def calculate_crc(data):
crc = sum(ord(ch) for ch in data) % 256
return f'{crc:03d}'
@staticmethod
def check_crc(data):
if data[-3:] != PfeifferProtocol.calculate_crc(data[:-3]):
raise CommunicationFailedError('Bad crc')
def communicate(self, cmd):
cmd += self.calculate_crc(cmd)
reply = super().communicate(cmd)
self.check_crc(reply)
return reply[:-3]
class PfeifferMixin(HasIO):
ioClass = PfeifferProtocol
address= Property('Addresse', datatype= IntRange())
def calculate_crc(self, data):
crc = sum(ord(chr) for chr in data) % 256
return f'{crc:03d}'
def check_crc(self, data):
if data [-3:] != self.calculate_crc(data[:-3]):
raise CommunicationFailedError('Bad crc')
def data_request_u_expo_new(self, parameter_nr):
cmd = f'{self.address:03d}00{parameter_nr:03d}02=?'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
reply = self.communicate(f'{self.address:03d}00{parameter_nr:03d}02=?')
assert int(reply[5:8]) == parameter_nr
@ -40,23 +44,16 @@ class PfeifferMixin(HasIO):
exponent = int(reply[14:16])-23
except ValueError:
raise CommunicationFailedError(f'got {reply[10:16]}')
return float(f'{reply[10:14]}e{exponent}')
def data_request_old_boolean(self, parameter_nr):
cmd = f'{self.address:03d}00{parameter_nr:03d}02=?'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
reply = self.communicate(f'{self.address:03d}00{parameter_nr:03d}02=?')
assert int(reply[5:8]) == parameter_nr, f"Parameter number mismatch: expected {parameter_nr}, got {int(reply[5:8])}"
assert int(reply[0:3]) == self.address, f"Address mismatch: expected {self.address}, got {int(reply[0:3])}"
if reply[12] == "1":
value = True
elif reply[12] == "0":
@ -67,12 +64,8 @@ class PfeifferMixin(HasIO):
return value
def data_request_u_real(self, parameter_nr):
cmd = f'{self.address:03d}00{parameter_nr:03d}02=?'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
reply = self.communicate(f'{self.address:03d}00{parameter_nr:03d}02=?')
assert int(reply[5:8]) == parameter_nr
assert int(reply[0:3]) == self.address
@ -83,19 +76,14 @@ class PfeifferMixin(HasIO):
raise CommunicationFailedError(f'got {reply[10:16]}')
return value
def data_request_u_int(self, parameter_nr):
cmd = f'{self.address:03d}00{parameter_nr:03d}02=?'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
reply = self.communicate(f'{self.address:03d}00{parameter_nr:03d}02=?')
if reply[8] == "0":
reply_length = (int)(reply[9])
reply_length = int(reply[9])
else:
reply_length = (int)(reply[8:10])
reply_length = int(reply[8:10])
try:
if reply[10 : 10 + reply_length] == "000000":
@ -108,31 +96,24 @@ class PfeifferMixin(HasIO):
return value
def data_request_string(self, parameter_nr):
cmd = f'{self.address:03d}00{parameter_nr:03d}02=?'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
reply = self.communicate(f'{self.address:03d}00{parameter_nr:03d}02=?')
assert int(reply[5:8]) == parameter_nr
assert int(reply[0:3]) == self.address
return str(reply[10:16])
def control_old_boolean(self, parameter_nr, target):
if target:
val = 1
else:
val = 0
cmd = f'{self.address:03d}10{parameter_nr:03d}06{str(val)*6}'
cmd += self.calculate_crc(cmd)
reply = self.communicate(cmd)
self.check_crc(reply)
assert cmd == reply, f'got {reply} instead of {cmd} '
try:
@ -145,25 +126,29 @@ class PfeifferMixin(HasIO):
raise CommunicationFailedError(f'got {reply[10:16]}')
return value
class RPT200(PfeifferMixin, Readable):
value = Parameter('Pressure', FloatRange(unit='hPa'))
def read_value(self):
return self.data_request_u_expo_new(740)
def read_status(self):
errtxt = self.data_request_string(303)
if errtxt == "000000":
return IDLE, ''
else:
return ERROR, errtxt
class TCP400(PfeifferMixin, Drivable, Readable):
speed= Parameter('Rotational speed', FloatRange(unit = 'Hz'), readonly = False)
target= Parameter('Pumping station', BoolType())
current= Parameter('Current consumption', FloatRange(unit = '%'))
value = Parameter('Turbopump state', BoolType())
temp = Parameter('temp', FloatRange(unit = 'C'))
def read_temp (self):
return self.data_request_u_int(326)
@ -184,6 +169,9 @@ class TCP400(PfeifferMixin, Drivable, Readable):
def read_status(self):
if not self.data_request_old_boolean(306):
return BUSY, 'ramping up'
if self.target:
return BUSY, 'ramping up'
else:
return IDLE, ''
else:
return IDLE,'at targetspeed'
return IDLE, 'at targetspeed'