144 lines
5.3 KiB
Python
144 lines
5.3 KiB
Python
#!/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())
|