195 lines
7.1 KiB
Python
195 lines
7.1 KiB
Python
"""
|
|
ptlogger -- host-side library for the 12-channel PT1000 temperature logger.
|
|
|
|
Wraps the binary USB-CDC protocol implemented in Application/cmd.c. All multi-byte
|
|
values are little-endian. The device shows up as a virtual COM port (CDC ACM).
|
|
|
|
Example
|
|
-------
|
|
from ptlogger import PTLogger
|
|
|
|
with PTLogger("COM5") as pt:
|
|
print(pt.version())
|
|
for ch, (temp, state) in enumerate(pt.read_all()):
|
|
print(ch, temp, pt.state_name(state))
|
|
"""
|
|
|
|
import json
|
|
import struct
|
|
import time
|
|
|
|
import serial # pyserial
|
|
import serial.tools.list_ports
|
|
|
|
NUM_CHANNELS = 12
|
|
|
|
# Sentinel returned by the firmware in place of a temperature when a channel
|
|
# is not OK (see PT_INVALID_MILLIC in max31865.h).
|
|
TEMP_INVALID = 0x7FFFFFFF
|
|
|
|
# Per-channel state codes (must match PT_State in max31865.h).
|
|
STATE_OK = 0 # connected, reading valid
|
|
STATE_OPEN = 1 # RTD open / no sensor connected
|
|
STATE_SHORT = 2 # RTD shorted
|
|
STATE_FAULT = 3 # other MAX31865 fault (REFIN / RTDIN / over-/under-voltage)
|
|
STATE_COMERR = 4 # SPI failure or invalid channel
|
|
|
|
STATE_NAMES = {
|
|
STATE_OK: "ok",
|
|
STATE_OPEN: "open",
|
|
STATE_SHORT: "short",
|
|
STATE_FAULT: "fault",
|
|
STATE_COMERR: "comerr",
|
|
}
|
|
|
|
|
|
class PTError(Exception):
|
|
"""Raised on a protocol / communication error (e.g. response timeout)."""
|
|
|
|
|
|
def list_ports():
|
|
"""Return a list of (device, description) tuples for the available serial ports."""
|
|
return [(p.device, p.description) for p in serial.tools.list_ports.comports()]
|
|
|
|
|
|
def state_name(code):
|
|
"""Human-readable name for a state code."""
|
|
return STATE_NAMES.get(code, "unknown(%d)" % code)
|
|
|
|
|
|
class PTLogger:
|
|
"""Connection to one temperature-logger board."""
|
|
|
|
def __init__(self, port, timeout=1.0, baudrate=115200):
|
|
# baudrate is ignored by a USB-CDC device but pyserial requires a value.
|
|
self.ser = serial.Serial(port, baudrate=baudrate, timeout=timeout)
|
|
time.sleep(0.1) # let the CDC link settle
|
|
self.ser.reset_input_buffer()
|
|
|
|
# -- context manager -------------------------------------------------
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
|
|
def close(self):
|
|
if self.ser.is_open:
|
|
self.ser.close()
|
|
|
|
# -- low-level -------------------------------------------------------
|
|
def _cmd(self, payload, nresp):
|
|
"""Send a command and read exactly `nresp` response bytes."""
|
|
self.ser.reset_input_buffer() # drop any stale bytes -> stay in sync
|
|
self.ser.write(payload)
|
|
data = self.ser.read(nresp)
|
|
if len(data) != nresp:
|
|
raise PTError("timeout: expected %d bytes, got %d" % (nresp, len(data)))
|
|
return data
|
|
|
|
@staticmethod
|
|
def _to_celsius(milli_c, state):
|
|
if state != STATE_OK or milli_c == TEMP_INVALID:
|
|
return None
|
|
return milli_c / 1000.0
|
|
|
|
@staticmethod
|
|
def state_name(code):
|
|
"""Human-readable name for a state code."""
|
|
return state_name(code)
|
|
|
|
# -- temperature -----------------------------------------------------
|
|
def read_temp(self, ch):
|
|
"""Read one channel. Returns (temperature_degC_or_None, state)."""
|
|
if not 0 <= ch < NUM_CHANNELS:
|
|
raise ValueError("channel out of range: %d" % ch)
|
|
milli_c, state = struct.unpack("<iB", self._cmd(bytes([ord("T"), ch]), 5))
|
|
return self._to_celsius(milli_c, state), state
|
|
|
|
def read_all(self):
|
|
"""Read all 12 channels. Returns a list of (temperature_degC_or_None, state)."""
|
|
data = self._cmd(b"A", NUM_CHANNELS * 5)
|
|
out = []
|
|
for i in range(NUM_CHANNELS):
|
|
milli_c, state = struct.unpack_from("<iB", data, i * 5)
|
|
out.append((self._to_celsius(milli_c, state), state))
|
|
return out
|
|
|
|
def read_raw(self, ch):
|
|
"""Read the raw 15-bit RTD value of one channel. Returns (raw15, state)."""
|
|
if not 0 <= ch < NUM_CHANNELS:
|
|
raise ValueError("channel out of range: %d" % ch)
|
|
raw, state = struct.unpack("<HB", self._cmd(bytes([ord("R"), ch]), 3))
|
|
return raw, state
|
|
|
|
# -- diagnostics -----------------------------------------------------
|
|
def read_status(self):
|
|
"""Return a list of 12 state codes (one per channel)."""
|
|
return list(self._cmd(b"S", NUM_CHANNELS))
|
|
|
|
def connected_mask(self):
|
|
"""Return a 16-bit mask; bit n set means channel n is a working PT1000."""
|
|
(mask,) = struct.unpack("<H", self._cmd(b"C", 2))
|
|
return mask
|
|
|
|
def connected_channels(self):
|
|
"""Return the list of channel indices that are connected and OK."""
|
|
mask = self.connected_mask()
|
|
return [ch for ch in range(NUM_CHANNELS) if mask & (1 << ch)]
|
|
|
|
# -- zero calibration ------------------------------------------------
|
|
def zero(self):
|
|
"""Zero-calibrate: the firmware measures all channels and stores a per-channel
|
|
offset so every connected channel then reads their common mean. Use with all
|
|
sensors in one homogeneous bath. Returns (count, mean_degC_or_None)."""
|
|
count, mean = struct.unpack("<Bi", self._cmd(b"Z", 5))
|
|
return count, (mean / 1000.0 if count else None)
|
|
|
|
def get_offsets(self):
|
|
"""Return the 12 current calibration offsets in degrees Celsius."""
|
|
data = self._cmd(b"O", NUM_CHANNELS * 4)
|
|
return [struct.unpack_from("<i", data, i * 4)[0] / 1000.0 for i in range(NUM_CHANNELS)]
|
|
|
|
def set_offsets(self, offsets_c):
|
|
"""Set all 12 offsets (list of degrees Celsius). All-zero clears the calibration."""
|
|
if len(offsets_c) != NUM_CHANNELS:
|
|
raise ValueError("need exactly %d offsets" % NUM_CHANNELS)
|
|
payload = b"P" + b"".join(struct.pack("<i", int(round(o * 1000.0))) for o in offsets_c)
|
|
self._cmd(payload, 1) # firmware acks with 1 byte
|
|
|
|
def clear_zero(self):
|
|
"""Remove the calibration (set all offsets to 0)."""
|
|
self.set_offsets([0.0] * NUM_CHANNELS)
|
|
|
|
def save_offsets(self, path):
|
|
"""Read the current offsets and store them to a JSON file. Returns the offsets."""
|
|
offsets = self.get_offsets()
|
|
with open(path, "w") as f:
|
|
json.dump(offsets, f)
|
|
return offsets
|
|
|
|
def load_offsets(self, path):
|
|
"""Load offsets from a JSON file and apply them (restores a saved calibration)."""
|
|
with open(path) as f:
|
|
offsets = json.load(f)
|
|
self.set_offsets(offsets)
|
|
return offsets
|
|
|
|
# -- info ------------------------------------------------------------
|
|
def version(self):
|
|
"""Return the firmware version string."""
|
|
self.ser.reset_input_buffer()
|
|
self.ser.write(b"V")
|
|
n = self.ser.read(1)
|
|
if len(n) != 1:
|
|
raise PTError("timeout reading version length")
|
|
text = self.ser.read(n[0])
|
|
if len(text) != n[0]:
|
|
raise PTError("timeout reading version string")
|
|
return text.decode("ascii", "replace")
|
|
|
|
def error(self):
|
|
"""Return the global error word (0 = no error)."""
|
|
(err,) = struct.unpack("<H", self._cmd(b"E", 2))
|
|
return err
|