diff options
Diffstat (limited to 'mpm/python/n3xx_bist')
| -rwxr-xr-x | mpm/python/n3xx_bist | 697 |
1 files changed, 135 insertions, 562 deletions
diff --git a/mpm/python/n3xx_bist b/mpm/python/n3xx_bist index 18f1c3b4b..cc7dd40c6 100755 --- a/mpm/python/n3xx_bist +++ b/mpm/python/n3xx_bist @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright 2017 Ettus Research, National Instruments Company +# Copyright 2017-2018 Ettus Research, National Instruments Company # # SPDX-License-Identifier: GPL-3.0-or-later # @@ -11,17 +11,9 @@ Will work on all derivatives of the N3xx series. """ from __future__ import print_function -import os import sys -import subprocess -import re -import socket -import select import time -import json -from datetime import datetime -import argparse -from six import iteritems +from usrp_mpm import bist # Timeout values are in seconds: GPS_WARMUP_TIMEOUT = 70 # Data sheet says "about a minute" @@ -31,105 +23,20 @@ GPS_LOCKOK_TIMEOUT = 2 # Data sheet says about 15 minutes. Because our test # by default. ############################################################################## -# Aurora/SFP BIST code +# Bist class ############################################################################## -def get_sfp_bist_defaults(): - " Default dictionary for SFP/Aurora BIST dry-runs " - return { - 'elapsed_time': 1.0, - 'max_roundtrip_latency': 0.8e-6, - 'throughput': 1000e6, - 'max_ber': 8.5e-11, - 'errors': 0, - 'bits': 12012486656, - } - -def assert_aurora_image(master, slave): - """ - Make sure we have an FPGA image with which we can run the requested tests. - - Will load an AA image if not, which always satisfies all conditions for - running Aurora tests. - """ - from usrp_mpm.sys_utils import uio - if not uio.find_uio_device(master)[0] or \ - (slave is not None and not uio.find_uio_device(slave)[0]): - load_fpga_image('AA') - -def run_aurora_bist(master, slave=None): +class N3XXBIST(bist.UsrpBIST): """ - Spawn a BER test - """ - from usrp_mpm import aurora_control - from usrp_mpm.sys_utils.uio import open_uio - - class DummyContext(object): - """Dummy class for context managers""" - def __enter__(self): - return - - def __exit__(self, exc_type, exc_value, traceback): - return exc_type is None - - # Go, go, go! - try: - assert_aurora_image(master, slave) - with open_uio(label=master, read_only=False) as master_au_uio: - master_au_ctrl = aurora_control.AuroraControl(master_au_uio) - with open_uio(label=slave, read_only=False)\ - if slave is not None else DummyContext() as slave_au_uio: - slave_au_ctrl = aurora_control.AuroraControl(slave_au_uio)\ - if slave is not None else None - return master_au_ctrl.run_ber_loopback_bist( - duration=10, - requested_rate=1300 * 8e6, - slave=slave_au_ctrl, - ) - except Exception as ex: - print("Unexpected exception: {}".format(str(ex))) - exit(1) - - -def aurora_results_to_status(bist_results): - """ - Convert a dictionary coming from AuroraControl BIST to one that we can use - for this BIST + BIST Tool for the USRP N3xx series """ - return bist_results['mst_errors'] == 0, { - 'elapsed_time': bist_results['time_elapsed'], - 'max_roundtrip_latency': bist_results['mst_latency_us'], - 'throughput': bist_results['approx_throughput'], - 'max_ber': bist_results['max_ber'], - 'errors': bist_results['mst_errors'], - 'bits': bist_results['mst_samps'], + usrp_type = "N3XX" + # This defines special tests that are really collections of other tests. + collections = { + 'standard': ["ddr3", "gpsdo", "rtc", "temp", "fan", "tpm"], + 'extended': "*", } - -############################################################################## -# Helpers -############################################################################## -def post_results(results): - """ - Given a dictionary, post the results. - - This will print the results as JSON to stdout. - """ - print(json.dumps( - results, - sort_keys=True, - indent=4, - separators=(',', ': ') - )) - -def filter_results_for_lv(results): - """ - The LabView JSON parser does not support a variety of things, such as - nested dicts, and some downstream LV applications freak out if certain keys - are not what they expect. - This is a long hard-coded list of how results should look like for those - cases. Note: This list needs manual supervision and attention for the case - where either subsystems get renamed, or other architectural changes should - occur. - """ + # Default FPGA image type + DEFAULT_FPGA_TYPE = 'HG' lv_compat_format = { 'ddr3': { 'throughput': -1, @@ -184,240 +91,24 @@ def filter_results_for_lv(results): 'lock_status': 0, }, } - # OK now go and brush up the results: - def fixup_dict(result_dict, ref_dict): - """ - Touches up result_dict according to ref_dict by the following rules: - - If a key is in result_dict that is not in ref_dict, delete that - - If a key is in ref_dict that is not in result_dict, use the value - from ref_dict - """ - ref_dict['error_msg'] = "" - ref_dict['status'] = False - result_dict = { - k: v for k, v in iteritems(result_dict) - if k in ref_dict or k in ('error_msg', 'status') - } - result_dict = { - k: result_dict.get(k, ref_dict[k]) for k in ref_dict - } - return result_dict - results = { - testname: fixup_dict(testresults, lv_compat_format[testname]) \ - if testname in lv_compat_format else testresults - for testname, testresults in iteritems(results) - } - return results - -def sock_read_line(my_sock, timeout=60, interval=0.1): - """ - Read from a socket until newline. If there was no newline until the timeout - occurs, raise an error. Otherwise, return the line. - """ - line = b'' - end_time = time.time() + timeout - while time.time() < end_time: - socket_ready = select.select([my_sock], [], [], 0)[0] - if socket_ready: - next_char = my_sock.recv(1) - if next_char == b'\n': - return line.decode('ascii') - line += next_char - else: - time.sleep(interval) - raise RuntimeError("sock_read_line() exceeded read timeout!") - -def poll_with_timeout(state_check, timeout_ms, interval_ms): - """ - Calls state_check() every interval_ms until it returns a positive value, or - until a timeout is exceeded. - - Returns True if state_check() returned True within the timeout. - """ - max_time = time.time() + (float(timeout_ms) / 1000) - interval_s = float(interval_ms) / 1000 - while time.time() < max_time: - if state_check(): - return True - time.sleep(interval_s) - return False - -def expand_options(option_list): - """ - Turn a list ['foo=bar', 'spam=eggs'] into a dictionary {'foo': 'bar', - 'spam': 'eggs'}. - """ - return dict(x.split('=') for x in option_list) - -############################################################################## -# Bist class -############################################################################## -class N3XXBIST(object): - """ - BIST Tool for the USRP N3xx series - """ - # This defines special tests that are really collections of other tests. - collections = { - 'standard': ["ddr3", "gpsdo", "rtc", "temp", "fan", "tpm"], - 'extended': "*", - } - # Default FPGA image type - DEFAULT_FPGA_TYPE = 'HG' - - @staticmethod - def make_arg_parser(): - """ - Return arg parser - """ - parser = argparse.ArgumentParser( - description="N3xx BIST Tool", - ) - parser.add_argument( - '-n', '--dry-run', action='store_true', - help="Fake out the tests. All tests will return a valid" \ - " response, but will not actually interact with hardware.", - ) - parser.add_argument( - '-v', '--verbose', action='store_true', - help="Crank up verbosity level", - ) - parser.add_argument( - '--debug', action='store_true', - help="For debugging this tool.", - ) - parser.add_argument( - '--option', '-o', action='append', default=[], - help="Option for individual test.", - ) - parser.add_argument( - '--lv-compat', action='store_true', - help="Provides compatibility with the LV JSON parser. Don't run " - "this mode unless you know what you're doing. The JSON " - "output does not necessarily reflect the actual system " - "status when using this mode.", - ) - parser.add_argument( - '--skip-fpga-reload', action='store_true', - help="Skip reloading the default FPGA image post-test. Note: by" - "specifying this argument, the FPGA image loaded could be " - "anything post-test.", - ) - parser.add_argument( - 'tests', - help="List the tests that should be run", - nargs='+', # There has to be at least one - ) - return parser + device_args = "type=n3xx,addr=127.0.0.1" def __init__(self): - self.args = N3XXBIST.make_arg_parser().parse_args() - self.args.option = expand_options(self.args.option) - # If this is true, trigger a reload of the default FPGA image - self.reload_fpga_image = False - try: - from usrp_mpm.periph_manager.n3xx import n3xx - default_rev = n3xx.mboard_max_rev - except ImportError: - # This means we're in dry run mode or something like that, so just - # pick something - default_rev = 3 - self.mb_rev = int(self.args.option.get('mb_rev', default_rev)) - self.tests_to_run = set() - for test in self.args.tests: - if test in self.collections: - for test in self.expand_collection(test): - self.tests_to_run.add(test) - else: - self.tests_to_run.add(test) - try: - # Keep this import here so we can do dry-runs without any MPM code - from usrp_mpm import get_main_logger - if not self.args.verbose: - from usrp_mpm.mpmlog import WARNING - get_main_logger().setLevel(WARNING) - self.log = get_main_logger().getChild('main') - except ImportError: - print("No logging capability available.") - - def expand_collection(self, coll): - """ - Return names of tests in a collection - """ - tests = self.collections[coll] - if tests == "*": - tests = {x.replace('bist_', '') - for x in dir(self) - if x.find('bist_') == 0 - } - else: - tests = set(tests) - return tests - - def run(self): - """ - Execute tests. + bist.UsrpBIST.__init__(self) - Returns True on Success. - """ - def execute_test(testname): - """ - Actually run a test. - """ - testmethod_name = "bist_{0}".format(testname) - sys.stderr.write( - "Executing test method: {0}\n\n".format(testmethod_name) - ) - try: - status, data = getattr(self, testmethod_name)() - data['status'] = status - data['error_msg'] = data.get('error_msg', '') - return status, data - except AttributeError: - sys.stderr.write("Test not defined: {}\n".format(testname)) - return False, {} - except Exception as ex: - sys.stderr.write( - "Test {} failed to execute: {}\n".format(testname, str(ex)) - ) - if self.args.debug: - raise - return False, {'error_msg': str(ex)} - tests_successful = True - result = {} - for test in self.tests_to_run: - status, result_data = execute_test(test) - tests_successful = tests_successful and status - result[test] = result_data - if self.args.lv_compat: - result = filter_results_for_lv(result) - post_results(result) - if self.reload_fpga_image and not self.args.skip_fpga_reload: - load_fpga_image(self.DEFAULT_FPGA_TYPE) - return tests_successful + def get_mb_periph_mgr(self): + """Return reference to an n3xx periph manager""" + from usrp_mpm.periph_manager.n3xx import n3xx + return n3xx + + def get_product_id(self): + """Return the mboard product ID (n310 or n300):""" + return bist.get_product_id_from_eeprom(valid_ids=['n300', 'n310']) ############################################################################# # BISTS # All bist_* methods must return True/False success values! ############################################################################# - def bist_rtc(self): - """ - BIST for RTC (real time clock) - - Return dictionary: - - date: Returns the current UTC time, with seconds-accuracy, in ISO 8601 - format, as a string. As if running 'date -Iseconds -u'. - - time: Same time, but in seconds since epoch. - - Return status: - Unless datetime throws an exception, returns True. - """ - assert 'rtc' in self.tests_to_run - utc_now = datetime.utcnow() - return True, { - 'time': time.mktime(utc_now.timetuple()), - 'date': utc_now.replace(microsecond=0).isoformat() + "+00:00", - } - def bist_ddr3(self): """ BIST for PL DDR3 DRAM @@ -437,25 +128,7 @@ class N3XXBIST(object): assert 'ddr3' in self.tests_to_run if self.args.dry_run: return True, {'throughput': 1250e6} - result = {} - ddr3_bist_executor = 'uhd_usrp_probe --args addr=127.0.0.1' - try: - output = subprocess.check_output( - ddr3_bist_executor, - stderr=subprocess.STDOUT, - shell=True, - ) - except subprocess.CalledProcessError as ex: - # Don't throw errors from uhd_usrp_probe - output = ex.output - output = output.decode("utf-8") - mobj = re.search(r"Throughput: (?P<thrup>[0-9.]+)\s?MB", output) - if mobj is not None: - result['throughput'] = float(mobj.group('thrup')) * 1000 - else: - result['throughput'] = 0 - result['error_msg'] = result.get('error_msg', '') + \ - "\n\nFailed match throughput regex!" + result = bist.test_ddr3_with_usrp_probe() return result.get('throughput', 0) > 1000e3, result def bist_gpsdo(self): @@ -501,7 +174,7 @@ class N3XXBIST(object): sys.stderr.write( "Waiting for WARMUP to go low for up to {} seconds...\n".format( gps_warmup_timeout)) - if not poll_with_timeout( + if not bist.poll_with_timeout( lambda: not gpio_tca6424.get('GPS-WARMUP'), gps_warmup_timeout*1000, 1000 ): @@ -513,7 +186,7 @@ class N3XXBIST(object): sys.stderr.write( "Waiting for LOCKOK to go high for up to {} seconds...\n".format( gps_lockok_timeout)) - if not poll_with_timeout( + if not bist.poll_with_timeout( lambda: gpio_tca6424.get('GPS-LOCKOK'), gps_lockok_timeout*1000, 1000 @@ -528,24 +201,8 @@ class N3XXBIST(object): sys.stderr.write("GPS-ALARM status: {}\n".format( gpio_tca6424.get('GPS-ALARM') )) - # Now read back response from chip - my_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - my_sock.connect(('localhost', 2947)) - sys.stderr.write("Connected to GPSDO socket.\n") - query_cmd = b'?WATCH={"enable":true,"json":true}' - my_sock.sendall(query_cmd) - sys.stderr.write("Sent query: {}\n".format(query_cmd)) - sock_read_line(my_sock, timeout=10) - sys.stderr.write("Received initial newline.\n") - result = {} - while result.get('class', None) != 'TPV': - json_result = sock_read_line(my_sock, timeout=60) - sys.stderr.write( - "Received JSON response: {}\n\n".format(json_result) - ) - result = json.loads(json_result) - my_sock.sendall(b'?WATCH={"enable":false}') - my_sock.close() + # Now the chip is on, read back the TPV result + result = bist.get_gpsd_tpv_result() # If we reach this line, we have a valid result and the chip responded. # However, it doesn't necessarily mean we had a GPS lock. return True, result @@ -567,17 +224,87 @@ class N3XXBIST(object): return True, { 'tpm0_caps': "Fake caps value\n\nVersion 0.0.0", } - result = {} - props_to_read = ('caps',) - base_path = '/sys/class/tpm' - for tpm_device in os.listdir(base_path): - if tpm_device.startswith('tpm'): - for key in props_to_read: - result['{}_{}'.format(tpm_device, key)] = open( - os.path.join(base_path, tpm_device, key), 'r' - ).read().strip() + result = bist.get_tpm_caps_info() return len(result) == 1, result + def bist_ref_clock_int(self): + """ + BIST for clock lock from internal (25 MHz) source. + Description: Checks to see if the daughtercard can lock to an internal + clock source. + + External Equipment: None + Return dictionary: + - <sensor-name>: + - locked: Boolean lock status + + There can be multiple ref lock sensors; for a pass condition they all + need to be asserted. + """ + assert 'ref_clock_int' in self.tests_to_run + if self.args.dry_run: + return True, {'ref_locked': True} + result = bist.get_ref_clock_prop( + 'internal', + 'internal', + extra_args={'skip_rfic': 1, 'rfnoc_num_blocks': 0} + ) + return 'error_msg' not in result, result + + def bist_ref_clock_ext(self): + """ + BIST for clock lock from external source. Note: This test requires a + connected daughterboard with a 'ref lock' sensor available. + + Description: Checks to see if the daughtercard can lock to the external + reference clock. + + External Equipment: 10 MHz reference Source connected to "ref in". + + Return dictionary: + - <sensor-name>: + - locked: Boolean lock status + + There can be multiple ref lock sensors; for a pass condition they all + need to be asserted. + """ + assert 'ref_clock_ext' in self.tests_to_run + if self.args.dry_run: + return True, {'ref_locked': True} + result = bist.get_ref_clock_prop( + 'external', + 'external', + extra_args={'skip_rfic': 1, 'rfnoc_num_blocks': 0} + ) + return 'error_msg' not in result, result + + def bist_ref_clock_gpsdo(self): + """ + BIST for clock lock from external source. Note: This test requires a + connected daughterboard with a 'ref lock' sensor available. + + Description: Checks to see if the daughtercard can lock to the external + reference clock. + + External Equipment: 10 MHz reference Source connected to "ref in". + + Return dictionary: + - <sensor-name>: + - locked: Boolean lock status + + There can be multiple ref lock sensors; for a pass condition they all + need to be asserted. + """ + assert 'ref_clock_gpsdo' in self.tests_to_run + if self.args.dry_run: + return True, {'ref_locked': True} + result = bist.get_ref_clock_prop( + 'gpsdo', + 'gpsdo', + extra_args={'skip_rfic': 1, 'rfnoc_num_blocks': 0} + ) + return 'error_msg' not in result, result + def bist_sfp0_loopback(self): """ BIST for SFP+ ports: @@ -596,10 +323,14 @@ class N3XXBIST(object): - bits: Number of bits that were transferred """ if self.args.dry_run: - return True, get_sfp_bist_defaults() - sfp_bist_results = run_aurora_bist(master='misc-auro-regs0') + return True, bist.get_sfp_bist_defaults() + sfp_bist_results = bist.run_aurora_bist( + device_args=self.device_args, + product_id=self.get_product_id(), + master='misc-auro-regs0', + ) self.reload_fpga_image = True - return aurora_results_to_status(sfp_bist_results) + return bist.aurora_results_to_status(sfp_bist_results) def bist_sfp1_loopback(self): """ @@ -619,10 +350,14 @@ class N3XXBIST(object): - bits: Number of bits that were transferred """ if self.args.dry_run: - return True, get_sfp_bist_defaults() - sfp_bist_results = run_aurora_bist(master='misc-auro-regs1') + return True, bist.get_sfp_bist_defaults() + sfp_bist_results = bist.run_aurora_bist( + device_args=self.device_args, + product_id=self.get_product_id(), + master='misc-auro-regs1', + ) self.reload_fpga_image = True - return aurora_results_to_status(sfp_bist_results) + return bist.aurora_results_to_status(sfp_bist_results) def bist_sfp_loopback(self): """ @@ -642,13 +377,15 @@ class N3XXBIST(object): - bits: Number of bits that were transferred """ if self.args.dry_run: - return True, get_sfp_bist_defaults() - sfp_bist_results = run_aurora_bist( + return True, bist.get_sfp_bist_defaults() + sfp_bist_results = bist.run_aurora_bist( + device_args=self.device_args, + product_id=self.get_product_id(), master='misc-auro-regs0', slave='misc-auro-regs1', ) self.reload_fpga_image = True - return aurora_results_to_status(sfp_bist_results) + return bist.aurora_results_to_status(sfp_bist_results) def bist_gpio(self): """ @@ -690,7 +427,7 @@ class N3XXBIST(object): " Run a GPIO test for a given set of patterns " gpio_ctrl = n3xx_periphs.FrontpanelGPIO(ddr) for pattern in patterns: - gpio_set_all(gpio_ctrl, pattern, GPIO_WIDTH, ddr) + bist.gpio_set_all(gpio_ctrl, pattern, GPIO_WIDTH, ddr) time.sleep(0.1) gpio_rb = gpio_ctrl.get_all() if pattern != gpio_rb: @@ -716,15 +453,8 @@ class N3XXBIST(object): assert 'temp' in self.tests_to_run if self.args.dry_run: return True, {'fpga-thermal-zone': 30000} - import pyudev - context = pyudev.Context() - result = { - device.attributes.get('type').decode('ascii'): \ - int(device.attributes.get('temp').decode('ascii')) - for device in context.list_devices(subsystem='thermal') - if 'temp' in device.attributes.available_attributes \ - and device.attributes.get('temp') is not None - } + result = bist.get_temp_sensor_value( + lambda device: device.attributes.get('type').decode('ascii')) if len(result) < 1: result['error_msg'] = "No temperature sensors found!" return 'error_msg' not in result, result @@ -742,14 +472,7 @@ class N3XXBIST(object): assert 'fan' in self.tests_to_run if self.args.dry_run: return True, {'cooling_device0': 10000, 'cooling_device1': 10000} - import pyudev - context = pyudev.Context() - result = { - device.sys_name: int(device.attributes.get('cur_state')) - for device in context.list_devices(subsystem='thermal') - if 'cur_state' in device.attributes.available_attributes \ - and device.attributes.get('cur_state') is not None - } + result = bist.get_fan_values() return len(result) == 2, result def bist_whiterabbit(self): @@ -768,7 +491,11 @@ class N3XXBIST(object): from usrp_mpm.sys_utils import uio if not uio.find_uio_device(n3xx.wr_regs_label, logger=self.log)[0]: self.log.info("Need to load WX image before proceeding...") - load_fpga_image('WX') + bist.load_fpga_image( + 'WX', + self.device_args, + self.get_product_id(), + ) self.log.info("Image loading complete.") self.reload_fpga_image = True mb_regs = n3xx_periphs.MboardRegsControl( @@ -776,7 +503,7 @@ class N3XXBIST(object): mb_regs.set_time_source('sfp0', 25e6) wr_regs_control = WhiteRabbitRegsControl( n3xx.wr_regs_label, self.log) - lock_status = poll_with_timeout( + lock_status = bist.poll_with_timeout( lambda: wr_regs_control.get_time_lock_status(), 40000, # Try for x ms... this number is set from a few benchtop tests 1000, # Poll every... second! why not? @@ -786,160 +513,6 @@ class N3XXBIST(object): } return lock_status, result - def bist_ref_clock_int(self): - """ - BIST for clock and pps lock from internal (25MHz). - Description: Checks to see if we can lock to an internal - clock source. - - External Equipment: None - Return dictionary: - - <sensor-name>: - - locked: Boolean lock status - - There can be multiple ref lock sensors; for a pass condition they all - need to be asserted. - """ - assert 'ref_clock_int' in self.tests_to_run - result = self.ref_clock_helper('internal', 'internal') - return 'error_msg' not in result, result - - def bist_ref_clock_ext(self): - """ - BIST for clock pps lock from external (10MHz) source. - Description: Checks to see if we can lock to an external - clock source. - - External Equipment: External 10 MHz reference clock needed (from Octoclock) - Return dictionary: - - <sensor-name>: - - locked: Boolean lock status - - There can be multiple ref lock sensors; for a pass condition they all - need to be asserted. - """ - assert 'ref_clock_ext' in self.tests_to_run - result = self.ref_clock_helper('external', 'external') - return 'error_msg' not in result, result - - def bist_ref_clock_gpsdo(self): - """ - BIST for clock and pps lock from gpsdo (20MHz) source. - Description: Checks to see if we can lock to an external - clock source. - - External Equipment: None - Return dictionary: - - <sensor-name>: - - locked: Boolean lock status - - There can be multiple ref lock sensors; for a pass condition they all - need to be asserted. - """ - assert 'ref_clock_gpsdo' in self.tests_to_run - result = self.ref_clock_helper('gpsdo', 'gpsdo') - return 'error_msg' not in result, result - - def ref_clock_helper(self, clock_source, time_source): - """ - Helper function to determine reference clock lock - Description: Checks to see if we can lock to a clock source. - External Equipment: None - Return dictionary: - - <sensor-name>: - - locked: Boolean lock status - """ - assert clock_source in ("gpsdo", "internal", "external"),\ - "Invalid clock source selected ({}). Valid choices: {}".format( - clock_source, ("gpsdo", "internal", "external")) - - assert time_source in ("gpsdo", "internal", "external", "sfp0"),\ - "Invalid time source selected ({}). Valid choices: {}".format( - time_source, ("gpsdo", "internal", "external", "sfp0")) - - if self.args.dry_run: - return True, {'ref_locked': True} - result = {} - env = os.environ.copy() - env['UHD_LOG_CONSOLE_LEVEL'] = 'error' - cmd = ['uhd_usrp_probe', '--args', - 'addr=127.0.0.1,' - 'rfnoc_num_blocks=0,' - 'skip_rfic=1,' - 'clock_source={c},time_source={t}'.format(c=clock_source, - t=time_source), - '--sensor'] - sensor_path = '/mboards/0/sensors/ref_locked' - cmd.append(sensor_path) - ref_lock_executor = ' '.join(cmd) - try: - output = subprocess.check_output( - ref_lock_executor, - stderr=subprocess.STDOUT, - env=env, - shell=True, - ) - except subprocess.CalledProcessError as ex: - # Don't throw errors from uhd_usrp_probe - output = ex.output - output = output.decode("utf-8") - mobj = re.search(r"true$", output.strip()) - if mobj is not None: - result['ref_locked'] = True - else: - result['ref_locked'] = False - result['error_msg'] = ("Reference Clock not locked." - " Extra output:" + output) - return result - - - -def gpio_set_all(gpio_bank, value, gpio_size, ddr_mask): - """Helper function for set gpio. - What this function do is take decimal value and convert to a binary string - then try to set those individual bits to the gpio_bank. - Arguments: - gpio_bank -- gpio bank type. - value -- value to set onto gpio bank. - gpio_size -- size of the gpio bank - ddr_mask -- data direction register bit mask. 0 is input; 1 is output. - """ - ddr_size = bin(ddr_mask).count("1") - value_bitstring = ('{0:0' + str(ddr_size) + 'b}').format(value)[-(gpio_size):] - ddr_bitstring = ('{0:0' + str(gpio_size) + 'b}').format(ddr_mask)[-(gpio_size):] - for i in range(gpio_size): - if ddr_bitstring[gpio_size - 1 - i] == "1": - gpio_bank.set(i, value_bitstring[i % ddr_size]) - -def get_product_id(): - """Return the mboard product ID (n310 or n300):""" - cmd = ['eeprom-id'] - output = subprocess.check_output( - cmd, - stderr=subprocess.STDOUT, - shell=True, - ).decode('utf-8') - if 'n310' in output: - return 'n310' - elif 'n300' in output: - return 'n300' - raise AssertionError("Cannot determine product ID.") - -def load_fpga_image(fpga_type): - """Load an FPGA image (HG, XG, AA, ...)""" - cmd = ['uhd_image_loader', '--args', 'type=n3xx', '--fpga-path'] - images_folder = '/usr/share/uhd/images/' - fpga_file_name = \ - 'usrp_' + get_product_id() + '_fpga_' + fpga_type.upper() + '.bit' - fpga_image = images_folder + fpga_file_name - cmd.append(fpga_image) - cmd_str = ' '.join(cmd) - subprocess.check_output( - cmd_str, - stderr=subprocess.STDOUT, - shell=True - ) - ############################################################################## # main ############################################################################## |