init
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
*.csv
|
||||
__pycache__
|
||||
+143
@@ -0,0 +1,143 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Example logger for the 12-channel PT1000 board.
|
||||
|
||||
Reads all channels at a fixed interval and appends them to a CSV file with a
|
||||
timestamp. Stop any time by pressing Enter (recommended -- works reliably in the
|
||||
VS Code terminal) or Ctrl+C. The CSV is flushed after every row, so nothing is lost.
|
||||
|
||||
Usage
|
||||
-----
|
||||
python log_to_csv.py COM5 # 1 s interval, auto-named file
|
||||
python log_to_csv.py COM5 -i 5 -o run1.csv # every 5 s into run1.csv
|
||||
python log_to_csv.py --list # list available serial ports
|
||||
|
||||
A channel that is not connected (open) or faulty is written as an empty cell.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import datetime
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
from ptlogger import PTLogger, PTError, NUM_CHANNELS, state_name, list_ports
|
||||
|
||||
|
||||
def parse_args():
|
||||
ap = argparse.ArgumentParser(description="Log 12x PT1000 temperatures to CSV.")
|
||||
ap.add_argument("port", nargs="?", help="serial port, e.g. COM5 or /dev/ttyACM0")
|
||||
ap.add_argument("-o", "--output", default=None,
|
||||
help="CSV file (default: pt_log_<date_time>.csv)")
|
||||
ap.add_argument("-i", "--interval", type=float, default=1.0,
|
||||
help="sample interval in seconds (default: 1.0)")
|
||||
ap.add_argument("-z", "--zero", action="store_true",
|
||||
help="zero-calibrate before logging (all sensors must be in one "
|
||||
"homogeneous bath); stores per-channel offsets on the device")
|
||||
ap.add_argument("--list", action="store_true",
|
||||
help="list available serial ports and exit")
|
||||
return ap.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
|
||||
if args.list:
|
||||
ports = list_ports()
|
||||
if not ports:
|
||||
print("no serial ports found")
|
||||
for dev, desc in ports:
|
||||
print("%-12s %s" % (dev, desc))
|
||||
return 0
|
||||
|
||||
if not args.port:
|
||||
print("error: no port given (use --list to see available ports)", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
out = args.output or "pt_log_%s.csv" % datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
|
||||
try:
|
||||
pt = PTLogger(args.port)
|
||||
except Exception as e:
|
||||
print("error: cannot open %s: %s" % (args.port, e), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
with pt:
|
||||
try:
|
||||
print("firmware:", pt.version())
|
||||
except PTError as e:
|
||||
print("warning: could not read version:", e)
|
||||
|
||||
connected = pt.connected_channels()
|
||||
print("connected channels:", connected if connected else "none")
|
||||
|
||||
if args.zero:
|
||||
count, mean = pt.zero()
|
||||
if count:
|
||||
print("zeroed %d channels to mean %.3f degC" % (count, mean))
|
||||
print("offsets:", ["%.3f" % o for o in pt.get_offsets()])
|
||||
else:
|
||||
print("zero skipped: no channels connected")
|
||||
|
||||
# Stop the logger by pressing Enter (works in the VS Code terminal,
|
||||
# where Ctrl+C can kill the whole shell). Ctrl+C still works elsewhere.
|
||||
stop = threading.Event()
|
||||
if sys.stdin and sys.stdin.isatty():
|
||||
def _wait_for_enter():
|
||||
try:
|
||||
sys.stdin.readline()
|
||||
except Exception:
|
||||
pass
|
||||
stop.set()
|
||||
threading.Thread(target=_wait_for_enter, daemon=True).start()
|
||||
stop_hint = "press Enter (or Ctrl+C) to stop"
|
||||
else:
|
||||
stop_hint = "press Ctrl+C to stop"
|
||||
|
||||
with open(out, "w", newline="") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["timestamp"] + ["T%d_degC" % ch for ch in range(NUM_CHANNELS)])
|
||||
|
||||
print("logging to %s every %.3g s -- %s" % (out, args.interval, stop_hint))
|
||||
next_sample = time.monotonic()
|
||||
while not stop.is_set():
|
||||
timestamp = datetime.datetime.now().isoformat(timespec="milliseconds")
|
||||
try:
|
||||
readings = pt.read_all()
|
||||
except PTError as e:
|
||||
print("warning: read failed:", e)
|
||||
if stop.wait(args.interval):
|
||||
break
|
||||
continue
|
||||
|
||||
row = [timestamp] + ["" if temp is None else "%.3f" % temp
|
||||
for temp, _state in readings]
|
||||
writer.writerow(row)
|
||||
f.flush()
|
||||
|
||||
# live view on the console
|
||||
cells = []
|
||||
for temp, state in readings:
|
||||
cells.append(" ----" if temp is None else "%6.2f" % temp)
|
||||
print(timestamp, " ".join(cells))
|
||||
|
||||
# keep the cadence steady; wake early if the user pressed Enter
|
||||
next_sample += args.interval
|
||||
delay = next_sample - time.monotonic()
|
||||
if delay > 0:
|
||||
if stop.wait(delay):
|
||||
break
|
||||
else:
|
||||
next_sample = time.monotonic()
|
||||
print("\nstopped by user.")
|
||||
except KeyboardInterrupt:
|
||||
print("\nstopped by user.")
|
||||
|
||||
print("saved:", out)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+189
@@ -0,0 +1,189 @@
|
||||
"""
|
||||
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
|
||||
|
||||
# -- 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
|
||||
@@ -0,0 +1 @@
|
||||
pyserial>=3.5
|
||||
Reference in New Issue
Block a user