commit db392eaf52fe3cdce3a8bf5553fe46e573ee3f1d
Author: Noah Piqué <noah.pique@psi.ch>
Date:   Thu Jun 18 10:42:11 2026 +0200

    init

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..a9c44ae
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+*.csv
+__pycache__
\ No newline at end of file
diff --git a/log_to_csv.py b/log_to_csv.py
new file mode 100644
index 0000000..bc5f55e
--- /dev/null
+++ b/log_to_csv.py
@@ -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())
diff --git a/ptlogger.py b/ptlogger.py
new file mode 100644
index 0000000..5639aec
--- /dev/null
+++ b/ptlogger.py
@@ -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
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..7ad05ef
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1 @@
+pyserial>=3.5