summaryrefslogtreecommitdiff
path: root/bin/kaxxxxp-viewer
diff options
context:
space:
mode:
Diffstat (limited to 'bin/kaxxxxp-viewer')
-rwxr-xr-xbin/kaxxxxp-viewer358
1 files changed, 358 insertions, 0 deletions
diff --git a/bin/kaxxxxp-viewer b/bin/kaxxxxp-viewer
new file mode 100755
index 0000000..afceb57
--- /dev/null
+++ b/bin/kaxxxxp-viewer
@@ -0,0 +1,358 @@
+#!/usr/bin/env python3
+# vim:tabstop=4 softtabstop=4 shiftwidth=4 textwidth=160 smarttab expandtab colorcolumn=160
+#
+# Copyright (C) 2021 Daniel Friesel
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+"""kaxxxxp-viewer - Data Logger and Viewer for KAxxxxP power supplies
+
+DESCRIPTION
+
+kaxxxxp-viewer logs voltage and current readings provided by a KAxxxxP power supply.
+Measurements can be taken directly (by specifying <measurement duration> in seconds)
+or loaded from a logfile using --load <file>. Data can be plotted or aggregated on stdout.
+
+WARNING
+
+The power supply's serial communication protocol is supports both read and write
+operations. Communication errors or bugs may cause the power supply to set an
+incompatible voltage or current limit, which may result in damaged equipment or
+fire. By using this software, you acknowledge that you are aware of these risks
+and the following disclaimer.
+
+This software is provided by the copyright holders and contributors "as is" and
+any express or implied warranties, including, but not limited to, the implied
+warranties of merchantability and fitness for a particular purpose are
+disclaimed. In no event shall the copyright holder or contributors be liable
+for any direct, indirect, incidental, special, exemplary, or consequential
+damages (including, but not limited to, procurement of substitute goods or
+services; loss of use, data, or profits; or business interruption) however
+caused and on any theory of liability, whether in contract, strict liability,
+or tort (including negligence or otherwise) arising in any way out of the use
+of this software, even if advised of the possibility of such damage.
+
+OPTIONS
+"""
+
+import argparse
+import numpy as np
+import serial
+import serial.threaded
+import signal
+import sys
+import tempfile
+import time
+
+terminate_measurement = False
+
+
+def running_mean(x: np.ndarray, N: int) -> np.ndarray:
+ """
+ Compute `N` elements wide running average over `x`.
+
+ :param x: 1-Dimensional NumPy array
+ :param N: how many items to average. Should be even for optimal results.
+ """
+
+ # to ensure that output.shape == input.shape, we need to insert data
+ # at the boundaries
+ boundary_array = np.insert(x, 0, np.full((N // 2), x[0]))
+ boundary_array = np.append(boundary_array, np.full((N // 2 + N % 2 - 1), x[-1]))
+
+ return np.convolve(boundary_array, np.ones((N,)) / N, mode="valid")
+
+
+class SerialReader(serial.threaded.Protocol):
+ def __init__(self):
+ self.remaining_chars = 0
+ self.recv_buf = ""
+ self.lines = []
+
+ def expect(self, num_chars):
+ self.recv_buf = ""
+ self.remaining_chars = num_chars
+
+ def __call__(self):
+ return self
+
+ def data_received(self, data):
+ try:
+ str_data = data.decode("UTF-8")
+ self.recv_buf += str_data
+ except UnicodeDecodeError:
+ sys.stderr.write("UART output contains gargabe: {data}\n".format(data=data))
+
+ self.remaining_chars -= len(str_data)
+
+ if self.remaining_chars <= 0:
+ self.lines.append(self.recv_buf)
+
+ def get_expected_line(self):
+ if len(self.lines):
+ ret = self.lines[0]
+ self.lines = []
+ return ret
+ return None
+
+ def get_line(self):
+ if len(self.lines):
+ ret = self.lines[-1]
+ self.lines = []
+ return ret
+ return None
+
+
+class KA320:
+ def __init__(self, port):
+ self.ser = serial.serial_for_url(port, do_not_open=True)
+ self.ser.baudrate = 9600
+ self.ser.parity = "N"
+ self.ser.rtscts = False
+ self.ser.xonxoff = False
+
+ try:
+ self.ser.open()
+ except serial.SerialException as e:
+ sys.stderr.write(
+ "Could not open serial port {}: {}\n".format(self.ser.name, e)
+ )
+ sys.exit(1)
+
+ self.reader = SerialReader()
+ self.worker = serial.threaded.ReaderThread(self.ser, self.reader)
+ self.worker.start()
+
+ def rw(self, cmd, num_chars, exact=False):
+ self.reader.expect(num_chars)
+ self.ser.write(cmd)
+ time.sleep(0.1)
+ if exact:
+ return self.reader.get_expected_line()
+ return self.reader.get_line()
+
+ def connect(self):
+ return self.rw(b"*IDN?", 16)
+
+ def get_max_voltage(self):
+ return float(self.rw(b"VSET1?", 5))
+
+ def get_max_current(self):
+ return float(self.rw(b"ISET1?", 5, True))
+
+ def get_voltage(self):
+ try:
+ return float(self.rw(b"VOUT1?", 5))
+ except TypeError:
+ return None
+
+ def get_current(self):
+ try:
+ return float(self.rw(b"IOUT1?", 5, True))
+ except TypeError:
+ return None
+
+ def set_output(self, enable):
+ if enable:
+ self.ser.write(b"OUT1")
+ else:
+ self.ser.write(b"OUT0")
+ time.sleep(0.1)
+
+ def disconnect(self):
+ self.worker.stop()
+ self.ser.close()
+
+
+def graceful_exit(sig, frame):
+ global terminate_measurement
+ terminate_measurement = True
+
+
+def measure_data(port, filename, duration):
+ global terminate_measurement
+
+ signal.signal(signal.SIGINT, graceful_exit)
+ signal.signal(signal.SIGTERM, graceful_exit)
+ signal.signal(signal.SIGQUIT, graceful_exit)
+ korad = KA320(port)
+
+ start_ts = time.time()
+
+ if filename is not None:
+ output_handle = open(filename, "w+")
+ else:
+ output_handle = tempfile.TemporaryFile("w+")
+
+ print("# Device: " + korad.connect(), file=output_handle)
+ print("# Timestamp Voltage Current", file=output_handle)
+ while not terminate_measurement:
+ ts = time.time()
+ current = korad.get_current()
+ voltage = korad.get_voltage()
+ if voltage is not None and current is not None:
+ print(f"{ts:.3f} {voltage:5.2f} {current:5.3f}", file=output_handle)
+ elif voltage is not None:
+ print(f"{ts:.3f} {volvoltage:5.2f} NaN", file=output_handle)
+ elif current is not None:
+ print(f"{ts:.3f} NaN {current:5.3f}", file=output_handle)
+ else:
+ print(f"{ts:.3f} NaN NaN", file=output_handle)
+ time.sleep(0.1)
+
+ if duration and ts - start_ts > duration:
+ terminate_measurement = True
+
+ korad.disconnect()
+
+ output_handle.seek(0)
+ output = output_handle.read()
+ output_handle.close()
+
+ return output
+
+
+def plot_data(data, mode):
+ import matplotlib.pyplot as plt
+
+ if mode == "U":
+ (datahandle,) = plt.plot(data[:, 0], data[:, 1], "b-", label="U", markersize=1)
+ (meanhandle,) = plt.plot(
+ data[:, 0],
+ running_mean(data[:, 1], 10),
+ "r-",
+ label="mean(U, 10)",
+ markersize=1,
+ )
+ plt.legend(handles=[datahandle, meanhandle])
+ plt.ylabel("Voltage [V]")
+
+ elif mode == "I":
+ (datahandle,) = plt.plot(data[:, 0], data[:, 2], "b-", label="I", markersize=1)
+ (meanhandle,) = plt.plot(
+ data[:, 0],
+ running_mean(data[:, 2], 10),
+ "r-",
+ label="mean(I, 10)",
+ markersize=1,
+ )
+ plt.legend(handles=[datahandle, meanhandle])
+ plt.ylabel("Current [A]")
+
+ elif mode == "P":
+ (datahandle,) = plt.plot(
+ data[:, 0], data[:, 1] * data[:, 2], "b-", label="P", markersize=1
+ )
+ (meanhandle,) = plt.plot(
+ data[:, 0],
+ running_mean(data[:, 1] * data[:, 2], 10),
+ "r-",
+ label="mean(P, 10)",
+ markersize=1,
+ )
+ plt.legend(handles=[datahandle, meanhandle])
+ plt.ylabel("Power [W]")
+
+ plt.show()
+
+
+def parse_data(log_data, skip=None, limit=None):
+ lines = log_data.split("\n")
+ data_count = sum(map(lambda x: len(x) > 0 and x[0] != "#", lines))
+ data_lines = filter(lambda x: len(x) > 0 and x[0] != "#", lines)
+
+ data = np.empty((data_count, 3))
+ skip_index = 0
+ limit_index = data_count
+
+ for i, line in enumerate(data_lines):
+ fields = line.split()
+ if len(fields) == 3:
+ timestamp, voltage, current = map(float, fields)
+ else:
+ raise RuntimeError('cannot parse line "{}"'.format(line))
+
+ if i == 0:
+ first_timestamp = timestamp
+
+ timestamp = timestamp - first_timestamp
+
+ if skip is not None and timestamp < skip:
+ skip_index = i + 1
+ continue
+
+ if limit is not None and timestamp > limit:
+ limit_index = i - 1
+ break
+
+ data[i] = [timestamp, voltage, current]
+
+ data = data[skip_index:limit_index]
+
+ return data
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__
+ )
+ parser.add_argument("--load", metavar="FILE", type=str, help="Load data from FILE")
+ parser.add_argument(
+ "--port",
+ metavar="PORT",
+ type=str,
+ default="/dev/ttyACM0",
+ help="Set PSU serial port",
+ )
+ parser.add_argument(
+ "--save", metavar="FILE", type=str, help="Save measurement data in FILE"
+ )
+ parser.add_argument(
+ "--skip",
+ metavar="N",
+ type=float,
+ default=0,
+ help="Skip the first N seconds of data. This is useful to avoid startup code influencing the results of a long-running measurement",
+ )
+ parser.add_argument(
+ "--limit",
+ type=float,
+ metavar="N",
+ help="Limit analysis to the first N seconds of data",
+ )
+ parser.add_argument(
+ "--plot",
+ metavar="UNIT",
+ choices=["U", "I", "P"],
+ help="Plot voltage / current / power over time",
+ )
+ parser.add_argument(
+ "duration", type=int, nargs="?", help="Measurement duration in seconds"
+ )
+
+ args = parser.parse_args()
+
+ if args.load is None and args.duration is None:
+ print("Either --load or duration must be specified", file=sys.stderr)
+ sys.exit(1)
+
+ if args.load:
+ if args.load.endswith(".xz"):
+ import lzma
+
+ with lzma.open(args.load, "rt") as f:
+ log_data = f.read()
+ else:
+ with open(args.load, "r") as f:
+ log_data = f.read()
+ else:
+ log_data = measure_data(args.port, args.save, args.duration)
+
+ data = parse_data(log_data, skip=args.skip, limit=args.limit)
+
+ if args.plot:
+ plot_data(data, args.plot)
+
+
+if __name__ == "__main__":
+ main()