Utils: add a RaToPUS MoniMod test script

6e3d9a28 · Christos Gentsos · 08ed5bfb · 6e3d9a28
Commit 6e3d9a28 authored Oct 28, 2021 by Christos Gentsos
Hide whitespace changes
Inline Side-by-side

Showing with 174 additions and 0 deletions

run_ratopus_test.py utils/run_ratopus_test.py +174 -0

No files found.
--- a/utils/run_ratopus_test.py
+++ b/utils/run_ratopus_test.py
+#!/bin/env python3
+
+import io, os
+import serial
+import argparse
+import time
+import math
+from datetime import datetime
+from datetime import timedelta
+from tabulate import tabulate
+from apscheduler.schedulers.background import BackgroundScheduler
+
+check_range_2s_30s = lambda x: check_range(2, 30, x)
+
+parser = argparse.ArgumentParser(description='Control the MoniMod test setup and log the results.')
+parser.add_argument('--log-interval', type=check_range_2s_30s, default=5,
+                    help='Logging interval (2s-30s, default=5).')
+
+parser.add_argument('--mm-log-file', type=argparse.FileType('w'),
+                    default=open('log_' + datetime.utcnow().strftime("%d.%m.%Y_%H:%M") + '_mm.txt', 'w'),
+                    help='MoniMod log file.')
+
+parser.add_argument('--mm-tty-dev', type=str,
+                    default='COM3',
+                    help='Test driver serial port device name.')
+
+args = parser.parse_args()
+
+def init_mm_tty():
+    global mm_tty
+    mm_tty = serial.Serial(args.mm_tty_dev, timeout=0.1)
+    mm_tty.close()
+    mm_tty.open()
+    if mm_tty.isOpen():
+        print(mm_tty.portstr, "Test driver TTY connection successful.")
+        return(0)
+    else:
+        return(-1)
+
+def reset_mm_controller():
+    print('resetting the test driver...')
+    mm_write('S')
+    mm_tty.close()
+    for i in range(5):
+        try:
+            print('Trying to reopen the test driver tty...')
+            time.sleep(2)
+            mm_tty.open()
+            print('Enabling MM test mode...')
+            mm_write('L')
+            time.sleep(0.1)
+            print('Enabling MM PEC...')
+            mm_write('E')
+            time.sleep(0.1)
+            print('Test driver successfully reset...')
+            break
+        except:
+            print('Failed to reopen the test driver tty...')
+
+def mm_write(cmd):
+    mm_tty.write(cmd.encode('utf-8'))
+
+def mm_read(cmd, timeout=1.5):
+    mm_tty.write(cmd.encode('utf-8'))
+    # while not mm_tty.in_waiting:
+    #     time.sleep(0.01)
+    time.sleep(timeout)
+    mm_tmp_line = mm_tty.read(256).decode('utf-8').strip().split('\n')
+    return(mm_tmp_line)
+
+def check_range(low, high, arg):
+    try:
+        value = int(arg)
+    except ValueError as err:
+        raise argparse.ArgumentTypeError(str(err))
+
+    if value < low or value > high:
+        message = "Expected {} <= value <= {}, got value = {}".format(low, high, value)
+        raise argparse.ArgumentTypeError(message)
+
+    return value
+
+mm_data_headers = ['Value', 'MM 0', 'MM 1']
+mm_data = [['N', 0],
+           ['TMR_ERR', 0],
+           ['I2C_RES', 0],
+           ['V1', 0.0],
+           ['V2', 0.0],
+           ['V3', 0.0],
+           ['I1', 0.0],
+           ['I2', 0.0],
+           ['I3', 0.0],
+           ['T1', 0.0],
+           ['T2', 0.0],
+           ['T3', 0.0],
+           ['F1', 0.0],
+           ['F2', 0.0],
+           ['F3', 0.0]]
+
+mm_error_msgs = []              # messages to print after the table
+
+def log_mm(f = None):
+    global mm_data
+    global mm_error_msgs
+
+    mm_ret = mm_read('a')
+
+    mm_ret = mm_ret.split('\t')
+
+    if int(mm_ret[1]) != 0:
+        mm_error_msgs += ['Misalignment in test driver response ({})'.format(mm_ret[1])]
+
+    mm_data[0][1] = int(mm_ret[0])
+    mm_data[1][1] = int(mm_ret[2])
+    mm_data[2][1] = int(mm_ret[3])
+    for i in range(12):
+        mm_data[i+3][1] = float(mm_ret[i+4])
+
+    if f:
+        f.write(datetime.utcnow().strftime("%d.%m.%Y %H:%M:%S.%f") + ',')
+        for j in range(0, 3):
+            f.write('{:d},'.format(mm_data[j][1]))
+        for j in range(3, 14):
+            f.write('{:.3f},'.format(mm_data[j][1]))
+        f.write('{:.3f}\n'.format(mm_data[14][1]))
+        f.flush()
+
+    if mm_data[2][1] != 0:
+        mm_error_msgs += ['I2C error ({})'.format(mm_data[2][1])]
+
+cnt = 0
+secs = 0
+
+def main_loop():
+    global cnt
+    global secs
+
+    mm_error_msgs = []
+
+    # skip first iteration
+    if cnt:
+        log_mm(args.mm_log_file)
+
+    os.system('clear')
+
+    print('Time elapsed: {}'.format(str(timedelta(seconds=secs))))
+    print()
+
+    print(tabulate(mm_data, mm_data_headers, floatfmt=".2f"), flush=True)
+
+    for a in mm_error_msgs:
+        print(a)
+
+    cnt = cnt + 1
+    secs = secs + args.log_interval
+
+
+def init():
+    init_mm_tty()
+    time.sleep(0.5)
+    reset_mm_controller()
+    time.sleep(1.0)
+
+# start
+init()
+
+sched = BackgroundScheduler()
+sched.add_job(main_loop, 'interval', seconds=args.log_interval)
+sched.start()
+
+while True:
+    None
+
+sched.shutdown()