#!/usr/bin/env python3 import re import os import sys import time import tempfile import argparse import subprocess import logging import datetime import pprint import inspect import json #------------------------------------------------------------------------------- class FormatterColored(logging.Formatter): """ Logging formatter with colored log levels and indented multi-line output """ def __init__(self, *args, **kwargs): logging.Formatter.__init__(self, *args, **kwargs) self.multi_line_indent = 45 self.color_map = { 'DEBUG' : 37, # white 'INFO' : 36, # cyan 'WARNING' : 33, # yellow 'ERROR' : 31, # red 'CRITICAL': 41, # white on red bg } def format(self, record): # Color and format levelname (if it isn't already colored) if not record.levelname[0:2] == '\033[': record.levelname = ('{}{}m{:>8}{}').format('\033[', self.color_map.get(record.levelname, 37), record.levelname, '\033[0m') # Indent multi-line log messages msg_lines = record.msg.split('\n') if len(msg_lines) > 1: record.msg = ('\n' + self.multi_line_indent*' ').join(msg_lines) return super().format(record) #------------------------------------------------------------------------------- class SampaTestError(RuntimeError): """ Exception base class for SAMPA tester (with error code) """ def __init__(self, message, error_code): super().__init__(message) self.error_code = error_code error_codes = { 'init': 10, 'deinit': 11, 'dft': 100, 'readout': 200, 'readall': 210, 'readnoise': 220, 'dc': 300, 'misc': 400, 'jtag': 500 } #------------------------------------------------------------------------------- class SubprocessWithLog(): """ Base class to facilitate execution of commands in subprocess and automated logging of the result to the scripts log file """ def __init__(self, logger, **kwargs): self.logger = logger self.error_id = kwargs.get("error_id") self.error_code = kwargs.get("error_code") class Error(SampaTestError): """ Helper class for error reporting """ def __init__(self, message, error_id, error_code): super().__init__(error_id + ": " + message, error_code) def execute_command(self, cmd, write_to_log = True): """ Execute single command in subprocess Returns full (subprocess) result object """ self.logger.debug('Executing \'' + ' '.join(x for x in cmd) + '\'') ret = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if (write_to_log): self.logger.debug(ret.stdout.decode()[:-1]) if (ret.returncode): self.logger.error('Non-zero return code ({:d}) for command \'{}\''.format(ret.returncode, ' '.join(x for x in cmd))) raise self.Error('Last command returned non-zero ({:d})'.format(ret.returncode), self.error_id, self.error_code) return ret def execute_commands(self, cmds, write_to_log = True): """ Wrapper to execute list of commands Returns list of result objects """ results = [] for cmd in cmds: results.append(self.execute_command(cmd, write_to_log)) return results #------------------------------------------------------------------------------- class DcMeasurements(SubprocessWithLog): """ DC measurements with external ADC """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.dc'), **{'error_code': error_codes['dc'], 'error_id': 'DC measurements'}) def run(self, *args, **kwargs): cmd = [ \ 'ttester', '-m', kwargs.get('mask', '0xf'), '--adc-all', '--json' ] result = self.execute_command(cmd) # FIXME: This handling of json stuff won't work with a mask with more than one bit set... maybe fix in ttester? result_json = json.loads(result.stdout.decode()[:-1]) # Values from the external ADC available as json object # FIXME: Do something with the values, compare, log to DB, whatever #------------------------------------------------------------------------------- class MiscMeasurements(SubprocessWithLog): """ Misc measurements """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.misc'), **{'error_code': error_codes['misc'], 'error_id': 'Misc'}) def run(self, *args, **kwargs): cmds = [\ ['ttester', '-m', kwargs.get('mask', '0xf'), '--test-mem'], ['ttester', '-m', kwargs.get('mask', '0xf'), '--test-sampa-i2c'], ['ttester', '-m', kwargs.get('mask', '0xf'), '--clk-config', hex(0x13)], ['ttester', '-m', kwargs.get('mask', '0xf'), '--clk-config'], ['ttester', '-m', kwargs.get('mask', '0xf'), '--clk-config', hex(0x41)], ['ttester', '-m', kwargs.get('mask', '0xf'), '--clk-config'], ['ttester', '-m', kwargs.get('mask', '0xf'), '--dump-sampa-registers'], ['ttester', '-m', kwargs.get('mask', '0xf'), '--test-ringosc'] ] result = self.execute_commands(cmds) #------------------------------------------------------------------------------- class JtagMeasurements(SubprocessWithLog): """ Jtag measurements """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.jtag'), **{'error_code': error_codes['jtag'], 'error_id': 'Jtag'}) def run(self, *args, **kwargs): setup = 0 if ( kwargs.get('mask', '0xf') == "0x2"): setup = 1 cmds = [\ ['ttester', '-m', kwargs.get('mask', '0xf'), '--dump-sampa-registers'], ['tsampa-jtag', '--setup', str(setup), '-a','-l',str(2)] ] result = self.execute_commands(cmds) #------------------------------------------------------------------------------- class Dft(SubprocessWithLog): def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.dft'), **{'error_code': error_codes['dft'], 'error_id': 'DFT test'}) def run(self, *args, **kwargs): # DFT test binary doesn't work with VLDB/FEC mask, but by specifying directly the setup to use # Use lowest set bit in mask as setup identifier mask = int(kwargs.get('mask', '0x1'), 0) setup = 0 for i in range(0, 32): if (mask >> i) & 0x1: setup = i break cmd = [ 'tsampa-dft', '--setup', str(setup), '--input-file', kwargs.get('input-file', '/local/data/dft/cov_90_logic.txt'), '-l', str(4) ] result = None try: result = self.execute_command(cmd) return result.returncode except SampaTestError as e: pass # FIXME: Ignore this for now, remove try/except when everything is fixed with DFT #------------------------------------------------------------------------------- class Readout(SubprocessWithLog): """ Perform readout of SAMPA under test (and run the SAMPA decoder? ...find the SYNC pattern? ...do other checks?) """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.read'), **{'error_code': error_codes['readout'], 'error_id': 'Readout test'}) def run(self, *args, **kwargs): cmd = [ \ 'treadout', '-m', kwargs.get('mask', '0xf'), '--nr', kwargs.get('nr', '0'), '--nr-prefix', kwargs.get('nr-prefix', 'sampa'), '--output-dir', kwargs.get('output-dir', '/tmp'), '--data-type', str(1), '--frames', str(300000), '--events', str(1), '--sampa-tester', '--overwrite' #FIXME: remove! ] if (kwargs.get('fn-postfix', '') != ''): cmd.append('--fn-postfix') cmd.append(kwargs.get('fn-postfix')) result = self.execute_command(cmd) #print(result.stdout.decode()) # FIXME: Do something with the recorded file here... throw some exception if stuff fails # replace bogus code in the following lines if (result.returncode != 0): self.logger.error('Readout test failed, see logfile: ') return result.returncode #------------------------------------------------------------------------------- class Readall(SubprocessWithLog): """ Perform full readout requence of SAMPA under test """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.readall'), **{'error_code': error_codes['readall'], 'error_id': 'Readall test'}) def run(self, *args, **kwargs): # set of runs with different pulser settings pulserString = [ "noise", "calib-odd", "calib-even", "delay-all"] pulserArg = [ "--none", "--odd", "--even", "--all" ] cg0String = [ "30mV", "20mV" ] cg0Arg = [ "0xff", "0x0" ] anaExe = [ "../ana/noise.sh", "../ana/calib.exe", "../ana/calib.exe", "../ana/measure_delay.exe" ] ana2Exe = [ "../ana/noise.exe", "../ana/measure_xtalk.exe", "../ana/measure_xtalk.exe", "../ana/bitcheck.exe" ] # for i in range(0,4): # communicate with pulse generator cmd = [ 'testpulser.sh', '-m', kwargs.get('mask', '0xf'), pulserArg[i] ] print(cmd) result = self.execute_command(cmd) for j in range(0,2): cmd = [ 'ttester', '-m', kwargs.get('mask', '0xf'), '--cg0', cg0Arg[j] ] result = self.execute_command(cmd) # take data fileString = cg0String[j] + "_" + pulserString[i] cmd = [ 'treadout', '-m', kwargs.get('mask', '0xf'), '--nr', kwargs.get('nr', '0'), '--nr-prefix', kwargs.get('nr-prefix', 'sampa'), '--output-dir', kwargs.get('output-dir', '/tmp'), '--data-type', str(1), '--frames', str(300000), '--events', str(1), '--fn-postfix', fileString, '--sampa-tester', '--overwrite' #FIXME: remove! ] print(cmd) result = self.execute_command(cmd) # decode data chipString = str(kwargs.get('nr', '0')).zfill(6) link = 0 if ( kwargs.get('mask', '0xf') == "0x2"): link = 2 binfilename = "{0}/{1}{2}/{1}{2}_trorc00_link0{3}_{4}.bin".format(kwargs.get('output-dir', '/tmp'), kwargs.get('nr-prefix', 'sampa'), chipString, link, fileString) adcfilename = "{0}/{1}{2}/{1}{2}_trorc00_link0{3}_{4}_adc.txt".format(kwargs.get('output-dir', '/tmp'), kwargs.get('nr-prefix', 'sampa'), chipString, link, fileString) cmd = [ 'sampa_raw_data_decoder', '-i', binfilename ] print(cmd) result = self.execute_command(cmd) # analyze data cmd = [ anaExe[i], chipString, fileString, adcfilename ] print(cmd) result = self.execute_command(cmd) # also do a second xtalk analysis for the calib files # if ((i==1) or (i==2)): # do a second analysis for all files if ( i<4 ): cmd = [ ana2Exe[i], chipString, fileString, adcfilename ] print(cmd) result = self.execute_command(cmd) # FIXME: Do something with the recorded file here... throw some exception if stuff fails # replace bogus code in the following lines if (result.returncode != 0): self.logger.error('Readout all test failed, see logfile: ') return result.returncode #------------------------------------------------------------------------------- class Readnoise(SubprocessWithLog): """ Perform noise readout requence of SAMPA under test """ def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.readnoise'), **{'error_code': error_codes['readnoise'], 'error_id': 'Readnoise test'}) def run(self, *args, **kwargs): # set of runs with different pulser settings pulserArg = "--none" cg0String = [ "30mV", "20mV" ] cg0Arg = [ "0xff", "0x0" ] anaExe = "../ana/noise.sh" ana2Exe = "../ana/noise.exe" # communicate with pulse generator cmd = [ 'testpulser.sh', '-m', kwargs.get('mask', '0xf'), pulserArg ] print(cmd) result = self.execute_command(cmd) for j in range(0,2): cmd = [ 'ttester', '-m', kwargs.get('mask', '0xf'), '--cg0', cg0Arg[j] ] print(cmd) result = self.execute_command(cmd) # take data fileString = cg0String[j] + "-reAna" # change _ to - from Readall method, and add "ReAna" cmd = [ 'treadout', '-m', kwargs.get('mask', '0xf'), '--nr', kwargs.get('nr', '0'), '--nr-prefix', kwargs.get('nr-prefix', 'sampa'), '--output-dir', kwargs.get('output-dir', '/tmp'), '--data-type', str(1), '--frames', str(300000), '--events', str(1), '--fn-postfix', fileString, '--sampa-tester', '--overwrite' #FIXME: remove! ] print(cmd) result = self.execute_command(cmd) # decode data chipString = str(kwargs.get('nr', '0')).zfill(6) link = 0 if ( kwargs.get('mask', '0xf') == "0x2"): link = 2 binfilename = "{0}/{1}{2}/{1}{2}_trorc00_link0{3}_{4}.bin".format(kwargs.get('output-dir', '/tmp'), kwargs.get('nr-prefix', 'sampa'), chipString, link, fileString) adcfilename = "{0}/{1}{2}/{1}{2}_trorc00_link0{3}_{4}_adc.txt".format(kwargs.get('output-dir', '/tmp'), kwargs.get('nr-prefix', 'sampa'), chipString, link, fileString) cmd = [ 'sampa_raw_data_decoder', '-i', binfilename ] print(cmd) result = self.execute_command(cmd) # analyze data cmd = [ anaExe, chipString, fileString, adcfilename ] print(cmd) result = self.execute_command(cmd) cmd = [ ana2Exe, chipString, fileString, adcfilename ] print(cmd) result = self.execute_command(cmd) # FIXME: Do something with the recorded file here... throw some exception if stuff fails # replace bogus code in the following lines if (result.returncode != 0): self.logger.error('Readout noise test failed, see logfile: ') return result.returncode #------------------------------------------------------------------------------- # Setup phase # class Setup(SubprocessWithLog): def __init__(self, *args, **kwargs): super().__init__(logging.getLogger('main.vldb'), **{'error_code': error_codes['init'], 'error_id': 'Setup phase'}) self.mask = kwargs.get('mask', '0xf') self.base_output_dir = kwargs.get('output-dir', '/tmp') self.nr_prefix = kwargs.get('nr-prefix', 'sampa') self.nr = int(kwargs.get('nr', '0'), 0) def initialize_trorc(self): cmds = [] for i in range(0,6): if (int(self.mask, 0) & (1 << i)): gbtrx_base_addr = 0xc000 + i*2*0x8000; # SAMPA tester: only even channels used cmds.extend([ \ # IDLE pattern ['trorc_rw', hex(gbtrx_base_addr + 0x1), hex(0xf0f0000f)], ['trorc_rw', hex(gbtrx_base_addr + 0x2), hex(0x000f0f00)], # SHR_OE disabled, BX_SYNC_TR enabled ['trorc_rw', hex(gbtrx_base_addr + 0x3), hex(0x00010000)], # DFT contoller configuration ['trorc_rw', hex(gbtrx_base_addr + 0x19), hex(0x00000001)] ]) return self.execute_commands(cmds) def dump_versions(self): r = self.execute_command(['trorc_status_dump'], False) self.logger.debug('\n'.join(r.stdout.decode().split('\n')[0:3])) # Only take first 3 lines of output (with version info) cmds = [\ ['ttester', '--version'], ['treadout', '--version'], ['tsampa-dft', '--version'], ['tsca', '--version'], ] return self.execute_commands(cmds) def initialize(self): """ Basic init routine """ cmds = [ \ ['tsca', '-m', self.mask, '--core-rst', '--core-init', '--svl-reset'], ['tsca', '-m', self.mask, '--sca-cfg-b', hex(0x3c), '--sca-cfg-c', hex(0x1), '--sca-cfg-d', hex(0x38), '--gpio-dir', hex(0x40000fff)], ['ttester', '-m', self.mask, '--mux-sel', hex(0x0)], ['ttester', '-m', self.mask, '--sampa-pwr', hex(0x1)], ['ttester', '-m', self.mask, '--sampa-rst'], ['ttester', '-m', self.mask, '--hadd', hex(0xf), '--clk-config', hex(0x41), '--cg0', hex(0x0), # 20 mV default '--cg1', hex(0xff), '--pol', hex(0xff), '--cts', hex(0x0)], ] return self.execute_commands(cmds) def initialize_readout(self): """ Initialize/prepare setup for readout test """ cmds = [ \ ['ttester', '-m', self.mask, '--clk-config', hex(0x41), '--mux-sel', hex(0x0)], ] return self.execute_commands(cmds) def initialize_misc(self): self.initialize_readout() def initialize_jtag(self): self.initialize_readout() def initialize_dft(self): """ Initialize setup for DFT test """ cmds = [ \ ['ttester', '-m', self.mask, '--clk-config', hex(0x13), '--mux-sel', hex(0x1)], ] return self.execute_commands(cmds) def deinitialize(self): cmds = [\ [ 'ttester', '-m', self.mask, '--sampa-pwr', '0x0'] ] return self.execute_commands(cmds) #------------------------------------------------------------------------------- def main(): #--------------------------------------- # Argument handling parser = argparse.ArgumentParser() parser.add_argument('-l', '--log-level', type=str, help='Specify log level', default='INFO') parser.add_argument('--logname', type=str, help='Specify logfile name', default='run.log') parser.add_argument('-m', '--mask', type=str, help='Specify VLDB board mask', required=True) parser.add_argument('--init', action='store_true', help='Basic initialization of setup') parser.add_argument('--test-dc', action='store_true', help='Gather external ADC measurements and compare') parser.add_argument('--test-misc', action='store_true', help='Misc other measurements') parser.add_argument('--test-jtag', action='store_true', help='Jtag measurements') parser.add_argument('--off', action='store_true', help='Turn off sampa-pwr') parser.add_argument('--test-dft', action='store_true', help='Run DFT test') parser.add_argument('--test-readout', action='store_true', help='Run readout test') parser.add_argument('--test-readall', action='store_true', help='Run readout all') parser.add_argument('--test-readnoise', action='store_true', help='Run readout noise') # parser.add_argument('--nr', type=str, help='Serial number of the SAMPA under test', required=True) parser.add_argument('--output-dir', default='/tmp', help='Base output directory') parser.add_argument('--nr-prefix', default='sampa', help='Specify default output prefix') args = parser.parse_args() #--------------------------------------- # Set up logger (console and file) formatter = FormatterColored(fmt='[%(name)10s][%(levelname)s] %(asctime)s | %(message)s', datefmt='%Y-%m-%d %H:%M:%S') logger = logging.getLogger('main') logger.setLevel(logging.DEBUG) slog = logging.StreamHandler() slog.setFormatter(formatter) slog.setLevel(args.log_level) logger.addHandler(slog) # Set up directory structure to dump all files (including log file ) output_path = os.path.join(args.output_dir, '{}{:06d}'.format(args.nr_prefix, int(args.nr, 0))) try: os.makedirs(output_path) except FileExistsError as e: if os.path.isfile(output_path): logger.fatal('{} already exists and is not a directory. Unable to proceed.'.format(output_path)) sys.exit(-1) else: logger.warn('Output directory {} already exists. Existing files might be overwritten.'.format(output_path)) flog = logging.FileHandler(os.path.join(output_path, args.logname), mode = 'w') flog.setFormatter(formatter) flog.setLevel(logging.DEBUG) logger.addHandler(flog) logger.debug('Command line arguments: {}'.format(str(args))) logger.info('Using output directory {}'.format(output_path)) #---------------------------------------------------------- # Do stuff try: s = Setup(**{'mask': args.mask}) if (args.init): logger.info('Initializing setup(s) {}'.format(args.mask)) s.dump_versions() s.initialize_trorc() s.initialize() logger.info('Setup(s) successfully initialized') if (args.off): logger.info('Turning off') s.deinitialize() logger.info('Turned off') if (args.test_dc): logger.info('Starting DC measurements (and comparison)') DcMeasurements().run(**{'mask': args.mask}) logger.info('DC measurements done') if (args.test_misc): s.initialize_misc() logger.info('Starting Misc measurements') MiscMeasurements().run(**{'mask': args.mask}) logger.info('Misc measurements done') if (args.test_jtag): s.initialize_jtag() logger.info('Starting Jtag measurements') JtagMeasurements().run(**{'mask': args.mask}) logger.info('Jtag measurements done') if (args.test_readout): s.initialize_readout() logger.info('Starting readout test') readout_params = {\ 'mask': args.mask, 'nr': args.nr, 'nr-prefix': args.nr_prefix, 'output-dir': args.output_dir, 'fn-postfix': 'test' } Readout().run(**readout_params) logger.info('Readout test done') if (args.test_readall): s.initialize_readout() logger.info('Starting readall test') readall_params = {\ 'mask': args.mask, 'nr': args.nr, 'nr-prefix': args.nr_prefix, 'output-dir': args.output_dir, 'fn-postfix': 'test' } Readall().run(**readall_params) logger.info('Readall test done') if (args.test_readnoise): s.initialize_readout() logger.info('Starting readnoise test') readnoise_params = {\ 'mask': args.mask, 'nr': args.nr, 'nr-prefix': args.nr_prefix, 'output-dir': args.output_dir, 'fn-postfix': 'test' } Readnoise().run(**readnoise_params) logger.info('Readnoise test done') if (args.test_dft): s.initialize_dft() logger.info('Starting DFT test') Dft().run(**{'mask': args.mask}) logger.info('Dft test done') except SampaTestError as e: logger.fatal('Detected exception: Error code {:d}, {}'.format(e.error_code, e)) sys.exit(e.error_code) #------------------------------------------------------------------------------- if __name__ == "__main__": main()