brewpi/components/sensor/tempSensor_max31865.py

#!/usr/bin/env python3
import time
from components import ATemperatureSensor, TemperatureSensorException
import spidev


class TempSensor_max31865(ATemperatureSensor):
	def name(self):
		return "Max31865"

	def __init__(self, temp_offset=0, **kwargs):
		ATemperatureSensor.__init__(self)
		self.temp_offset = temp_offset
		# Set before the first read attempt so a read failure even on the
		# very first call (temperature() below, or TempSensorTask's own
		# priming call) has a well-defined last-known-good value to fall
		# back to rather than raising AttributeError.
		self.temp = None
		self.spi = spidev.SpiDev()
		try:
			self._open_spi()
		except Exception as e:
			# Don't crash the whole server if the SPI bus is momentarily
			# busy/unavailable at startup (e.g. right after a Pi reboot) -
			# temperature()'s own except handler already retries via
			# reopen() on the next tick, same recovery path as a later
			# runtime failure.
			print(f"TempSensor_max31865: could not open SPI at startup: {e}")

	def _open_spi(self):
		self.spi.open(0, 0)
		self.spi.max_speed_hz = 50000
		self.spi.mode = 0b01
		self.write_reg(0x00, 0xA3)

	def reopen(self):
		"""Closes and reopens the SPI handle - recovers from a wedged bus/
		chip-select state (e.g. left mid-transaction by some other fault)
		without needing a full process or Pi reboot. There's no serial-
		style connect/disconnect concept for a bus peripheral like this
		one, so this is the closest equivalent - called automatically by
		temperature() below on a read failure.

		Never raises - it's itself the recovery path called from
		temperature()'s except handler, and if it raised too that would
		escape temperature() entirely (same class of bug as HendiCtrl.
		disconnect() previously letting a comm failure escape and
		permanently kill HeaterTask's coroutine - see tasks/heater.py's
		on_process()). A genuinely still-broken bus just leaves self.temp
		at its last-known-good value, logged, and gets retried again next
		tick."""
		try:
			self.spi.close()
		except Exception:
			pass
		try:
			self._open_spi()
		except Exception as e:
			print(f"TempSensor_max31865: reopen failed: {e}")

	def read_reg(self, addr):
		reg = self.spi.xfer([addr, 0xFF])

		return reg[1]

	def write_reg(self, addr, data):
		self.spi.xfer([addr+0x80, data])

	def read_digits(self):
		for i in range(0, 10):
			reg = self.read_reg(0x00)
			if (0x20 & reg) == 0:
				break
			time.sleep(0.01)

		digits = 0
		msb = self.read_reg(0x01)
		lsb = self.read_reg(0x02)
		self.write_reg(0x00, 0xA3)
		if lsb & 0x01 == 0x00:
			digits = (256*msb + lsb) >> 1
		else:
			raise TemperatureSensorException("Could not read sensor!")

		return digits


	def temperature(self):
		try:
			digits = self.read_digits()
			self.temp = self.temp_offset + self.to_temperature(digits)
		except Exception as e:
			# Holds the last-known-good self.temp rather than raising -
			# TempSensorTask.on_process() calls this every tick with no
			# guard of its own (mirrors HeaterHendi.process()'s pattern of
			# self-healing at the actor level rather than letting a comm
			# failure escape and permanently kill the task's coroutine -
			# nothing restarts a dead ATask). reopen() gives the SPI bus a
			# chance to recover from a wedged state without a full process/
			# Pi reboot - see docs/pi_deployment_notes.md.
			print(f"TempSensor_max31865: comm error reading sensor, reopening SPI: {e}")
			self.reopen()
		return self.temp

	@staticmethod
	def to_temperature(digits):
		T = 0
		R = __class__.calc_R(430, digits)
		T = __class__.vanDusen_temp(R)
		
		return T

	@staticmethod
	def calc_R(Rref, digits):
		k = float(digits) / 32768
		R = k*Rref

		return R

	@staticmethod
	def vanDusen_temp(Rmeas):
		RLut, TLut = __class__.vanDusenLut(100, -150, +500, 1.0)
				
		# Find candidate
		i = 0
		for R in RLut:
			if R > Rmeas:
				break
			i += 1

		# Linear interpolation
		R0 = RLut[i-1]
		R1 = RLut[i]
		T0 = TLut[i-1]
		T1 = TLut[i]

		dR = R1 - R0
		dT = T1 - T0

		k = (Rmeas-R0)/dR
		T = T0 + dT*k

		return T

	@staticmethod
	def vanDusenLut (R0, Tmin, Tmax, dT):
		a = +3.90830e-03
		b = -5.77500e-07
		c = -4.18301e-12

		Rv = []
		Tv = []
		T = Tmin
		while T <= Tmax:
			R = R0*(1 + a*T + b*T**2)

			if T < 0.0:
				R += R0*c*(T - 100)*T**3

			Rv.append(R)
			Tv.append(T)
			T += dT

		return Rv, Tv

# Main
if __name__ == '__main__':


	sensor = TempSensor_max31865()
	for k in range (0, 100):
		for i in range (0, 8):
			print("Reg[0x{}]: 0x{:02X}".format(i, sensor.read_reg(i)))
		
		digits = sensor.read_digits()
		print("Digits = {:04X}".format(digits))
		print("Temperature = {:0.2f} °C".format(TempSensor_max31865.to_temperature(digits)))
		time.sleep(0.5)

	print("End of program")