python3-damaris/src/damaris/tools/eurotherm.py

import serial
import re
import time
import operator
from functools import reduce

from .temperature import TemperatureController, register_controller


DEBUG=False

reply_pattern = re.compile(r"\x02..(.*)\x03.", re.DOTALL)
# example answer '\x02PV279.8\x03/'
# [EOT] = \x04
# [STX] = \x02
# [ENQ] = \x05
# [ETX] = \x03
# [ACK] = \x06
# BCC = checksum

standard_device='0011'
EOT = '\x04'
STX = '\x02'
ENQ = '\x05'
ETX = '\x03'
ACK = '\x06'
NAK = '\x15'


"""
Paramter read example:

Master: [EOT]0011PV[ENQ]
Instrument:				[STX]PV16.4[ETX]{BCC}

Writing data:

Master: [EOT] {GID}{GID}{UID}{UID}[STX]{CHAN}(c1)(c2)<DATA>[ETX](BCC)

"""

def checksum(message):
	bcc = (reduce(operator.xor, list(map(ord,message))))
	return chr(bcc)

class Eurotherm(TemperatureController):
	def __init__(self, serial_device, baudrate = 19200):
		self.device = standard_device
		# timeout: 110 ms to get all answers.
		self.s = serial.Serial(serial_device,
					baudrate = baudrate,
					bytesize=7,
					parity='E',
					stopbits=1,
					timeout=0.11)
		self._expect_len = 50

	def send_read_param(self, param):
		self.s.write(EOT + self.device + param + ENQ)

	def read_param(self, param, retries=3, retry_delay=0.05):
		for attempt in range(retries):
			self.s.flushInput()
			self.send_read_param(param)
			answer = self.s.read(self._expect_len)
			m = reply_pattern.search(answer)
			if m is None:
				answer += self.s.read(200)
				m = reply_pattern.search(answer)
			if m is not None:
				self._expect_len = len(answer)
				return m.group(1)
			if attempt < retries - 1:
				time.sleep(retry_delay)
		print("received:", repr(answer))
		return None

	def write_param(self, mnemonic, data):
		if len(mnemonic) > 2:
			raise ValueError
		bcc = checksum(mnemonic + data + ETX)
		mes = EOT+self.device+STX+mnemonic+data+ETX+bcc
		if DEBUG:
			for i in  mes:
				print(i,hex(ord(i)))
		self.s.flushInput()
		self.s.write(mes)
		answer = self.s.read(1)
		# print "received:", repr(answer)
		if answer == "":
			# raise IOError("No answer from device")
			return None
		return answer[-1] == ACK

	def get_current_temperature(self):
		temp = self.read_param('PV')
		if temp is None:
			temp = "0"
		return temp

	def set_temperature(self, temperature):
		return self.write_param('SL', str(temperature))

	def get_setpoint_temperature(self):
		return self.read_param('SL')

	# TemperatureController interface implementation

	def get_temperature(self) -> float:
		"""Read current temperature in Celsius."""
		temp = self.read_param('PV')
		if temp is None:
			return 0.0
		try:
			return float(temp)
		except (ValueError, TypeError):
			return 0.0

	def get_setpoint(self) -> float:
		"""Read target setpoint temperature in Celsius."""
		temp = self.read_param('SL')
		if temp is None:
			return 0.0
		try:
			return float(temp)
		except (ValueError, TypeError):
			return 0.0

	def set_setpoint(self, temperature: float) -> bool:
		"""Set target temperature in Celsius."""
		return self.write_param('SL', str(temperature))

	def check_connection(self) -> bool:
		"""Check if the Eurotherm device is connected and responsive."""
		try:
			result = self.read_param('PV')
			return result is not None
		except Exception:
			return False

	def close(self):
		"""Close the serial connection."""
		if hasattr(self, 's') and self.s is not None:
			if self.s.is_open:
				self.s.close()
			self.s = None

if __name__ == '__main__':
	import time
	delta=5
	date = time.strftime('%Y-%m-%d')
	f = open('templog_%s'%date,'w')
	f.write('# Start time: %s\n#delta t : %.1f s\n'%(time.asctime(), delta))
	et = Eurotherm("/dev/ttyUSB0")
	while True:
		for i in range(120):
			time.sleep(delta)
			#t = time.strftime()
			T = et.get_current_temperature()
				
			l = '%f %s\n'%(time.time(),T)
			print(time.asctime(), T)
			f.write(l)
			f.flush()
		f.write('# MARK -- %s --\n'%(time.asctime()))

# Register Eurotherm with the factory
register_controller('eurotherm', Eurotherm)