saymrwulf/uhd
1
0
Fork 0
mirror of https://github.com/saymrwulf/uhd.git synced 2026-05-16 21:10:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
uhd/mpm/python/usrp_mpm/periph_manager/x4xx.py
Martin Braun 57e8c028d8 mpm: x410: Fix docstrings related to GPS sensors 
No code changes, only comment changes.
2022-01-10 08:02:51 -06:00

1347 lines
55 KiB
Python
Raw Blame History

#
# Copyright 2019 Ettus Research, a National Instruments Company
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
"""
X400 implementation module
"""
import threading
import copy
from time import sleep
from os import path
from collections import namedtuple
from usrp_mpm import lib # Pulls in everything from C++-land
from usrp_mpm import tlv_eeprom
from usrp_mpm.cores import WhiteRabbitRegsControl
from usrp_mpm.components import ZynqComponents
from usrp_mpm.sys_utils import dtoverlay
from usrp_mpm.sys_utils import ectool
from usrp_mpm.sys_utils import i2c_dev
from usrp_mpm.sys_utils.gpio import Gpio
from usrp_mpm.sys_utils.udev import dt_symbol_get_spidev
from usrp_mpm.rpc_server import no_claim, no_rpc
from usrp_mpm.mpmutils import assert_compat_number, poll_with_timeout
from usrp_mpm.periph_manager import PeriphManagerBase
from usrp_mpm.xports import XportMgrUDP
from usrp_mpm.periph_manager.x4xx_periphs import MboardRegsControl
from usrp_mpm.periph_manager.x4xx_periphs import CtrlportRegs
from usrp_mpm.periph_manager.x4xx_dio_control import DioControl
from usrp_mpm.periph_manager.x4xx_periphs import QSFPModule
from usrp_mpm.periph_manager.x4xx_periphs import get_temp_sensor
from usrp_mpm.periph_manager.x4xx_mb_cpld import MboardCPLD
from usrp_mpm.periph_manager.x4xx_clk_aux import ClockingAuxBrdControl
from usrp_mpm.periph_manager.x4xx_clk_mgr import X4xxClockMgr
from usrp_mpm.periph_manager.x4xx_gps_mgr import X4xxGPSMgr
from usrp_mpm.periph_manager.x4xx_rfdc_ctrl import X4xxRfdcCtrl
from usrp_mpm.dboard_manager.x4xx_db_iface import X4xxDboardIface
from usrp_mpm.dboard_manager.zbx import ZBX
X400_DEFAULT_EXT_CLOCK_FREQ = 10e6
X400_DEFAULT_MASTER_CLOCK_RATE = 122.88e6
X400_DEFAULT_TIME_SOURCE = X4xxClockMgr.TIME_SOURCE_INTERNAL
X400_DEFAULT_CLOCK_SOURCE = X4xxClockMgr.CLOCK_SOURCE_INTERNAL
X400_DEFAULT_ENABLE_PPS_EXPORT = True
X400_FPGA_COMPAT = (7, 3)
X400_DEFAULT_TRIG_DIRECTION = ClockingAuxBrdControl.DIRECTION_OUTPUT
X400_MONITOR_THREAD_INTERVAL = 1.0 # seconds
QSFPModuleConfig = namedtuple("QSFPModuleConfig", "modprs modsel devsymbol")
X400_QSFP_I2C_CONFIGS = [
QSFPModuleConfig(modprs='QSFP0_MODPRS', modsel='QSFP0_MODSEL_n', devsymbol='qsfp0_i2c'),
QSFPModuleConfig(modprs='QSFP1_MODPRS', modsel='QSFP1_MODSEL_n', devsymbol='qsfp1_i2c')]
RPU_SUCCESS_REPORT = 'Success'
RPU_FAILURE_REPORT = 'Failure'
RPU_REMOTEPROC_FIRMWARE_PATH = '/lib/firmware'
RPU_REMOTEPROC_PREFIX_PATH = '/sys/class/remoteproc/remoteproc'
RPU_REMOTEPROC_PROPERTY_FIRMWARE = 'firmware'
RPU_REMOTEPROC_PROPERTY_STATE = 'state'
RPU_STATE_COMMAND_START = 'start'
RPU_STATE_COMMAND_STOP = 'stop'
RPU_STATE_OFFLINE = 'offline'
RPU_STATE_RUNNING = 'running'
RPU_MAX_FIRMWARE_SIZE = 0x100000
RPU_MAX_STATE_CHANGE_TIME_IN_MS = 10000
RPU_STATE_CHANGE_POLLING_INTERVAL_IN_MS = 100
DIOAUX_EEPROM = "dioaux_eeprom"
DIOAUX_PID = 0x4003
# pylint: disable=too-few-public-methods
class EepromTagMap:
"""
Defines the tagmap for EEPROMs matching this magic.
The tagmap is a dictionary mapping an 8-bit tag to a NamedStruct instance.
The canonical list of tags and the binary layout of the associated structs
is defined in mpm/tools/tlv_eeprom/usrp_eeprom.h. Only the subset relevant
to MPM are included below.
"""
magic = 0x55535250
tagmap = {
# 0x10: usrp_eeprom_board_info
0x10: tlv_eeprom.NamedStruct('< H H H 7s 1x',
['pid', 'rev', 'rev_compat', 'serial']),
# 0x11: usrp_eeprom_module_info
0x11: tlv_eeprom.NamedStruct('< H H 7s 1x',
['module_pid', 'module_rev', 'module_serial']),
}
###############################################################################
# Transport managers
###############################################################################
class X400XportMgrUDP(XportMgrUDP):
"X400-specific UDP configuration"
iface_config = {
'sfp0': {
'label': 'misc-enet-regs0',
'type': 'sfp',
},
'sfp0_1': {
'label': 'misc-enet-regs0-1',
'type': 'sfp',
},
'sfp0_2': {
'label': 'misc-enet-regs0-2',
'type': 'sfp',
},
'sfp0_3': {
'label': 'misc-enet-regs0-3',
'type': 'sfp',
},
'sfp1': {
'label': 'misc-enet-regs1',
'type': 'sfp',
},
'sfp1_1': {
'label': 'misc-enet-regs1-1',
'type': 'sfp',
},
'sfp1_2': {
'label': 'misc-enet-regs1-2',
'type': 'sfp',
},
'sfp1_3': {
'label': 'misc-enet-regs1-3',
'type': 'sfp',
},
'int0': {
'label': 'misc-enet-int-regs',
'type': 'internal',
},
'eth0': {
'label': '',
'type': 'forward',
}
}
# pylint: enable=too-few-public-methods
###############################################################################
# Main Class
###############################################################################
class x4xx(ZynqComponents, PeriphManagerBase):
"""
Holds X400 specific attributes and methods
"""
#########################################################################
# Overridables
#
# See PeriphManagerBase for documentation on these fields. We try and keep
# them in the same order as they are in PeriphManagerBase for easier lookup.
#########################################################################
pids = {0x0410: 'x410'}
description = "X400-Series Device"
eeprom_search = PeriphManagerBase._EepromSearch.SYMBOL
# This is not in the overridables section from PeriphManagerBase, but we use
# it below
eeprom_magic = EepromTagMap.magic
mboard_eeprom_offset = 0
mboard_eeprom_max_len = 256
mboard_eeprom_magic = eeprom_magic
mboard_info = {"type": "x4xx"}
mboard_max_rev = 7 # RevG
max_num_dboards = 2
mboard_sensor_callback_map = {
# List of motherboard sensors that are always available. There are also
# GPS sensors, but they get added during __init__() only when there is
# a GPS available.
'ref_locked': 'get_ref_lock_sensor',
'fan0': 'get_fan0_sensor',
'fan1': 'get_fan1_sensor',
'temp_fpga' : 'get_fpga_temp_sensor',
'temp_internal' : 'get_internal_temp_sensor',
'temp_main_power' : 'get_main_power_temp_sensor',
'temp_scu_internal' : 'get_scu_internal_temp_sensor',
}
db_iface = X4xxDboardIface
dboard_eeprom_magic = eeprom_magic
updateable_components = {
'fpga': {
'callback': "update_fpga",
'path': '/lib/firmware/{}.bin',
'reset': True,
'check_dts_for_compatibility': True,
'compatibility': {
'fpga': {
'current': X400_FPGA_COMPAT,
'oldest': (7, 0),
},
'cpld_ifc' : {
'current': (2, 0),
'oldest': (2, 0),
},
'db_gpio_ifc': {
'current': (1, 0),
'oldest': (1, 0),
},
'rf_core_100m': {
'current': (1, 0),
'oldest': (1, 0),
},
'rf_core_400m': {
'current': (1, 0),
'oldest': (1, 0),
},
}
},
'dts': {
'callback': "update_dts",
'path': '/lib/firmware/{}.dts',
'output': '/lib/firmware/{}.dtbo',
'reset': False,
},
}
discoverable_features = ["ref_clk_calibration", "time_export", "trig_io_mode", "gpio_power"]
#
# End of overridables from PeriphManagerBase
###########################################################################
# X400-specific settings
# Label for the mboard UIO
mboard_regs_label = "mboard-regs"
ctrlport_regs_label = "ctrlport-mboard-regs"
# Label for the white rabbit UIO
wr_regs_label = "wr-regs"
# Override the list of updateable components
# X4xx specific discoverable features
@classmethod
def generate_device_info(cls, eeprom_md, mboard_info, dboard_infos):
"""
Hard-code our product map
"""
# Add the default PeriphManagerBase information first
device_info = super().generate_device_info(
eeprom_md, mboard_info, dboard_infos)
# Then add X4xx-specific information
mb_pid = eeprom_md.get('pid')
device_info['product'] = cls.pids.get(mb_pid, 'unknown')
module_serial = eeprom_md.get('module_serial')
if module_serial is not None:
device_info['serial'] = module_serial
return device_info
@staticmethod
def list_required_dt_overlays(device_info):
"""
Lists device tree overlays that need to be applied before this class can
be used. List of strings.
Are applied in order.
eeprom_md -- Dictionary of info read out from the mboard EEPROM
device_args -- Arbitrary dictionary of info, typically user-defined
"""
return [device_info['product']]
def _init_mboard_overlays(self):
"""
Load all required overlays for this motherboard
Overriden from the base implementation to force apply even if
the overlay was already loaded.
"""
requested_overlays = self.list_required_dt_overlays(
self.device_info,
)
self.log.debug("Motherboard requests device tree overlays: {}".format(
requested_overlays
))
# Remove all overlays before applying new ones
for overlay in requested_overlays:
dtoverlay.rm_overlay_safe(overlay)
for overlay in requested_overlays:
dtoverlay.apply_overlay_safe(overlay)
# Need to wait here a second to make sure the ethernet interfaces are up
# TODO: Fine-tune this number, or wait for some smarter signal.
sleep(1)
###########################################################################
# Ctor and device initialization tasks
###########################################################################
def __init__(self, args):
super(x4xx, self).__init__()
self._tear_down = False
self._rpu_initialized = False
self._status_monitor_thread = None
self._master_clock_rate = None
self._gps_mgr = None
self._clk_mgr = None
self._safe_sync_source = {
'clock_source': X400_DEFAULT_CLOCK_SOURCE,
'time_source': X400_DEFAULT_TIME_SOURCE,
}
self.rfdc = None
self.mboard_regs_control = None
self.ctrlport_regs = None
self.cpld_control = None
self.dio_control = None
try:
self._init_peripherals(args)
self.init_dboards(args)
# We need to init dio_control separately from peripherals
# since it needs information about available dboards
self._init_dio_control(args)
self._clk_mgr.set_dboard_reset_cb(
lambda enable: [db.reset_clock(enable) for db in self.dboards])
except Exception as ex:
self.log.error("Failed to initialize motherboard: %s", str(ex), exc_info=ex)
self._initialization_status = str(ex)
self._device_initialized = False
if not self._device_initialized:
# Don't try and figure out what's going on. Just give up.
return
try:
if not args.get('skip_boot_init', False):
self.init(args)
except Exception as ex:
self.log.warning("Failed to initialize device on boot: %s", str(ex))
# Freeze the RFDC calibration by default
self.rfdc.set_cal_frozen(1, 1, "both")
self.rfdc.set_cal_frozen(1, 0, "both")
# The parent class versions of these functions require access to self, but
# these versions don't.
# pylint: disable=no-self-use
def _read_mboard_eeprom_data(self, eeprom_path):
""" Returns a tuple (eeprom_dict, eeprom_rawdata) for the motherboard
EEPROM.
"""
return tlv_eeprom.read_eeprom(eeprom_path, EepromTagMap.tagmap,
EepromTagMap.magic, None)
def _read_dboard_eeprom_data(self, eeprom_path):
""" Returns a tuple (eeprom_dict, eeprom_rawdata) for a daughterboard
EEPROM.
"""
return tlv_eeprom.read_eeprom(eeprom_path, EepromTagMap.tagmap,
EepromTagMap.magic, None)
# pylint: enable=no-self-use
def _check_fpga_compat(self):
" Throw an exception if the compat numbers don't match up "
actual_compat = self.mboard_regs_control.get_compat_number()
self.log.debug("Actual FPGA compat number: {:d}.{:d}".format(
actual_compat[0], actual_compat[1]
))
assert_compat_number(
X400_FPGA_COMPAT,
actual_compat,
component="FPGA",
fail_on_old_minor=False,
log=self.log
)
def _init_gps_mgr(self):
"""
Initialize the GPS manager and the sensors.
Note that mpmd_impl queries all available sensors at initialization
time, in order to populate the property tree. That means we can't
dynamically load/unload sensors. Instead, we have to make sure that
the sensors can handle the GPS sensors, even when it's disabled. That
is pushed into the GPS manager class.
"""
self.log.debug("Found GPS, adding sensors.")
gps_mgr = X4xxGPSMgr(self._clocking_auxbrd, self.log)
# We can't use _add_public_methods(), because we only want a subset of
# the public methods. Also, we want to know which sensors were added so
# we can also add them to mboard_sensor_callback_map.
new_methods = gps_mgr.extend(self)
self.mboard_sensor_callback_map.update(new_methods)
return gps_mgr
def _monitor_status(self):
"""
Status monitoring thread: This should be executed in a thread. It will
continuously monitor status of the following peripherals:
- REF lock (update back-panel REF LED)
"""
self.log.trace("Launching monitor loop...")
cond = threading.Condition()
cond.acquire()
while not self._tear_down:
ref_locked = self.get_ref_lock_sensor()['value'] == 'true'
if self._clocking_auxbrd is not None:
self._clocking_auxbrd.set_ref_lock_led(ref_locked)
# Now wait
if cond.wait_for(
lambda: self._tear_down,
X400_MONITOR_THREAD_INTERVAL):
break
cond.release()
self.log.trace("Terminating monitor loop.")
def _assert_rfdc_powered(self):
"""
Assert that RFdc power is enabled, throw RuntimeError otherwise.
"""
if not self._rfdc_powered.get():
err_msg = "RFDC is not powered on"
self.log.error(err_msg)
raise RuntimeError(err_msg)
def _get_serial_number(self):
"""
Read the serial number from eeprom, falling back to the board S/N
if the module S/N is not populated.
"""
serial_number = self._eeprom_head.get("module_serial")
if serial_number is None:
self.log.warning(
'Module serial number not programmed, falling back to motherboard serial')
serial_number = self._eeprom_head["serial"]
return serial_number.rstrip(b'\x00')
def _init_peripherals(self, args):
"""
Turn on all peripherals. This may throw an error on failure, so make
sure to catch it.
"""
# Sanity checks
assert self.mboard_info.get('product') in self.pids.values(), \
"Device product could not be determined!"
# Init peripherals
self._rfdc_powered = Gpio('RFDC_POWERED', Gpio.INPUT)
# Init RPU Manager
self.log.trace("Initializing RPU manager peripheral...")
self.init_rpu()
# Init clocking aux board
self.log.trace("Initializing Clocking Aux Board controls...")
has_gps = False
try:
self._clocking_auxbrd = ClockingAuxBrdControl()
self.log.trace("Initialized Clocking Aux Board controls")
has_gps = self._clocking_auxbrd.is_gps_supported()
except RuntimeError:
self.log.warning(
"GPIO I2C bus could not be found for the Clocking Aux Board, "
"disabling Clocking Aux Board functionality.")
self._clocking_auxbrd = None
self._safe_sync_source = {
'clock_source': X4xxClockMgr.CLOCK_SOURCE_MBOARD,
'time_source': X4xxClockMgr.TIME_SOURCE_INTERNAL,
}
initial_clock_source = args.get('clock_source', X400_DEFAULT_CLOCK_SOURCE)
if self._clocking_auxbrd:
self._add_public_methods(self._clocking_auxbrd, "clkaux")
else:
initial_clock_source = X4xxClockMgr.CLOCK_SOURCE_MBOARD
# Init CPLD before talking to clocking ICs
cpld_spi_node = dt_symbol_get_spidev('mb_cpld')
self.cpld_control = MboardCPLD(cpld_spi_node, self.log)
self.cpld_control.check_signature()
self.cpld_control.check_compat_version()
self.cpld_control.trace_git_hash()
self._assert_rfdc_powered()
# Init clocking after CPLD as the SPLL communication is relying on it.
# We try and guess the correct master clock rate here based on defaults
# and args. Since we are still in __init__(), the args that come from mpm.conf
# are empty. We can't detect the real default MCR, because we need
# the RFDC controls for that -- but they won't work without clocks. So
# let's pick a sensible default MCR value, init the clocks, and fix the
# MCR value further down.
self._master_clock_rate = float(
args.get('master_clock_rate', X400_DEFAULT_MASTER_CLOCK_RATE))
sample_clock_freq, _, is_legacy_mode, _ = \
X4xxRfdcCtrl.master_to_sample_clk[self._master_clock_rate]
self._clk_mgr = X4xxClockMgr(
initial_clock_source,
time_source=args.get('time_source', X400_DEFAULT_TIME_SOURCE),
ref_clock_freq=float(args.get(
'ext_clock_freq', X400_DEFAULT_EXT_CLOCK_FREQ)),
sample_clock_freq=sample_clock_freq,
is_legacy_mode=is_legacy_mode,
clk_aux_board=self._clocking_auxbrd,
cpld_control=self.cpld_control,
log=self.log)
self._add_public_methods(
self._clk_mgr,
prefix="",
filter_cb=lambda name, method: not hasattr(method, '_norpc')
)
# Overlay must be applied after clocks have been configured
self.overlay_apply()
# Init Mboard Regs
self.log.trace("Initializing MBoard reg controls...")
serial_number = self._get_serial_number()
self.mboard_regs_control = MboardRegsControl(
self.mboard_regs_label, self.log)
self._check_fpga_compat()
self.mboard_regs_control.set_serial_number(serial_number)
self.mboard_regs_control.get_git_hash()
self.mboard_regs_control.get_build_timestamp()
self._clk_mgr.mboard_regs_control = self.mboard_regs_control
# Create control for RFDC
self.rfdc = X4xxRfdcCtrl(self._clk_mgr.get_spll_freq, self.log)
self._add_public_methods(
self.rfdc, prefix="",
filter_cb=lambda name, method: not hasattr(method, '_norpc')
)
self._update_fpga_type()
# Force reset the RFDC to ensure it is in a good state
self.rfdc.set_reset(reset=True)
self.rfdc.set_reset(reset=False)
# Synchronize SYSREF and clock distributed to all converters
self.rfdc.sync()
self._clk_mgr.set_rfdc_reset_cb(self.rfdc.set_reset)
# The initial default mcr only works if we have an FPGA with
# a decimation of 2. But we need the overlay applied before we
# can detect decimation, and that requires clocks to be initialized.
self.set_master_clock_rate(self.rfdc.get_default_mcr())
# Init ctrlport endpoint
self.ctrlport_regs = CtrlportRegs(self.ctrlport_regs_label, self.log)
# Init IPass cable status forwarding and CMI
self.cpld_control.set_serial_number(serial_number)
self.cpld_control.set_cmi_device_ready(
self.mboard_regs_control.is_pcie_present())
# The CMI transmission can be disabled by setting the cable status
# to be not connected. All images except for the LV PCIe variant
# provide a fixed "cables are unconnected" status. The LV PCIe image
# reports the correct status. As the FPGA holds this information it
# is possible to always enable the iPass cable present forwarding.
self.ctrlport_regs.enable_cable_present_forwarding(True)
# Init QSFP modules
for idx, config in enumerate(X400_QSFP_I2C_CONFIGS):
attr = QSFPModule(
config.modprs, config.modsel, config.devsymbol, self.log)
setattr(self, "_qsfp_module{}".format(idx), attr)
self._add_public_methods(attr, "qsfp{}".format(idx))
# Init GPS
if has_gps:
self._gps_mgr = self._init_gps_mgr()
# Init CHDR transports
self._xport_mgrs = {
'udp': X400XportMgrUDP(self.log, args),
}
# Spawn status monitoring thread
self.log.trace("Spawning status monitor thread...")
self._status_monitor_thread = threading.Thread(
target=self._monitor_status,
name="X4xxStatusMonitorThread",
daemon=True,
)
self._status_monitor_thread.start()
# Init complete.
self.log.debug("Device info: {}".format(self.device_info))
def _init_dio_control(self, _):
"""
Turn on gpio peripherals. This may throw an error on failure, so make
sure to catch it.
"""
if self._check_compat_aux_board(DIOAUX_EEPROM, DIOAUX_PID):
self.dio_control = DioControl(self.mboard_regs_control,
self.cpld_control, self.log,
self.dboards)
# add dio_control public methods to MPM API
self._add_public_methods(self.dio_control, "dio")
def _check_compat_aux_board(self, name, pid):
"""
Check whether auxiliary board given by name and pid can be found
:param name: symbol name of the auxiliary board which is used as
lookup for the dictionary of available boards.
:param pid: PID the board must have to be considered compatible
:return True if board is available with matching PID,
False otherwise
"""
assert(isinstance(self._aux_board_infos, dict)), "No EEPROM data"
board_info = self._aux_board_infos.get(name, None)
if board_info is None:
self.log.warning("Board for %s not present" % name)
return False
if board_info.get("pid", 0) != pid:
self.log.error("Expected PID for board %s to be 0x%04x but found "
"0x%04x" % (name, pid, board_info["pid"]))
return False
self.log.debug("Found compatible board for %s "
"(PID: 0x%04x)" % (name, board_info["pid"]))
return True
def init_rpu(self):
"""
Initializes the RPU image manager
"""
if self._rpu_initialized:
return
# Check presence/state of RPU cores
try:
for core_number in [0, 1]:
self.log.trace(
"RPU Core %d state: %s",
core_number,
self.get_rpu_state(core_number))
# TODO [psisterh, 5 Dec 2019]
# Should we force core to
# stop if running or in error state?
self.log.trace("Initialized RPU cores successfully.")
self._rpu_initialized = True
except FileNotFoundError:
self.log.warning(
"Failed to initialize RPU: remoteproc sysfs not present.")
###########################################################################
# Session init and deinit
###########################################################################
def init(self, args):
"""
Calls init() on the parent class, and then programs the Ethernet
dispatchers accordingly.
"""
if not self._device_initialized:
self.log.warning(
"Cannot run init(), device was never fully initialized!")
return False
# We need to disable the PPS out during clock and dboard initialization in order
# to avoid glitches.
if self._clocking_auxbrd is not None:
self._clocking_auxbrd.set_trig(False)
# If the caller has not specified clock_source or time_source, set them
# to the values currently configured.
args['clock_source'] = args.get('clock_source', self._clk_mgr.get_clock_source())
args['time_source'] = args.get('time_source', self._clk_mgr.get_time_source())
self.set_sync_source(args)
# If a Master Clock Rate was specified,
# re-configure the Sample PLL and all downstream clocks
if 'master_clock_rate' in args:
self.set_master_clock_rate(float(args['master_clock_rate']))
# Initialize CtrlportRegs (manually opens the UIO resource for faster access)
self.ctrlport_regs.init()
# Note: The parent class takes care of calling init() on all the
# daughterboards
result = super(x4xx, self).init(args)
# Now the clocks are all enabled, we can also enable PPS export:
if self._clocking_auxbrd is not None:
self._clocking_auxbrd.set_trig(
args.get('pps_export', X400_DEFAULT_ENABLE_PPS_EXPORT),
args.get('trig_direction', X400_DEFAULT_TRIG_DIRECTION)
)
for xport_mgr in self._xport_mgrs.values():
xport_mgr.init(args)
return result
def deinit(self):
"""
Clean up after a UHD session terminates.
"""
if not self._device_initialized:
self.log.warning(
"Cannot run deinit(), device was never fully initialized!")
return
if self.get_ref_lock_sensor()['unit'] != 'locked':
self.log.error("ref clocks aren't locked, falling back to default")
source = {"clock_source": X400_DEFAULT_CLOCK_SOURCE,
"time_source": X400_DEFAULT_TIME_SOURCE
}
self.set_sync_source(source)
super(x4xx, self).deinit()
self.ctrlport_regs.deinit()
for xport_mgr in self._xport_mgrs.values():
xport_mgr.deinit()
def tear_down(self):
"""
Tear down all members that need to be specially handled before
deconstruction.
For X400, this means the overlay.
"""
self.log.trace("Tearing down X4xx device...")
self._tear_down = True
if self._device_initialized:
self._status_monitor_thread.join(3 * X400_MONITOR_THREAD_INTERVAL)
if self._status_monitor_thread.is_alive():
self.log.error("Could not terminate monitor thread! "
"This could result in resource leaks.")
# call tear_down on daughterboards first
super(x4xx, self).tear_down()
if self.dio_control is not None:
self.dio_control.tear_down()
self.rfdc.tear_down()
self._clk_mgr.unset_cbs()
# remove x4xx overlay
active_overlays = self.list_active_overlays()
self.log.trace("X4xx has active device tree overlays: {}".format(
active_overlays
))
for overlay in active_overlays:
dtoverlay.rm_overlay(overlay)
###########################################################################
# Transport API
###########################################################################
# pylint: disable=no-self-use
def get_chdr_link_types(self):
"""
This will only ever return a single item (udp).
"""
return ["udp"]
# pylint: enable=no-self-use
def get_chdr_link_options(self, xport_type):
"""
Returns a list of dictionaries. Every dictionary contains information
about one way to connect to this device in order to initiate CHDR
traffic.
The interpretation of the return value is very highly dependant on the
transport type (xport_type).
For UDP, the every entry of the list has the following keys:
- ipv4 (IP Address)
- port (UDP port)
- link_rate (bps of the link, e.g. 10e9 for 10GigE)
"""
if xport_type not in self._xport_mgrs:
self.log.warning("Can't get link options for unknown link type: `{}'.")
return []
if xport_type == "udp":
return self._xport_mgrs[xport_type].get_chdr_link_options(
self.mboard_info['rpc_connection'])
# else:
return self._xport_mgrs[xport_type].get_chdr_link_options()
###########################################################################
# Device info
###########################################################################
def get_device_info_dyn(self):
"""
Append the device info with current IP addresses.
"""
if not self._device_initialized:
return {}
device_info = self._xport_mgrs['udp'].get_xport_info()
device_info.update({
'fpga_version': "{}.{}".format(
*self.mboard_regs_control.get_compat_number()),
'fpga_version_hash': "{:x}.{}".format(
*self.mboard_regs_control.get_git_hash()),
'fpga': self.updateable_components.get('fpga', {}).get('type', ""),
})
return device_info
def is_db_gpio_ifc_present(self, slot_id):
"""
Return if daughterboard GPIO interface at 'slot_id' is present in the FPGA
"""
db_gpio_version = self.mboard_regs_control.get_db_gpio_ifc_version(slot_id)
return db_gpio_version[0] > 0
###########################################################################
# Clock/Time API
###########################################################################
def get_clock_sources(self):
"""
Lists all available clock sources.
"""
return self._clk_mgr.get_clock_sources()
def set_clock_source(self, *args):
"""
Ensures the new reference clock source and current time source pairing
is valid and sets both by calling set_sync_source().
"""
clock_source = args[0]
time_source = self._clk_mgr.get_time_source()
assert clock_source is not None
assert time_source is not None
if (clock_source, time_source) not in self._clk_mgr.valid_sync_sources:
old_time_source = time_source
if clock_source in (
X4xxClockMgr.CLOCK_SOURCE_MBOARD,
X4xxClockMgr.CLOCK_SOURCE_INTERNAL):
time_source = X4xxClockMgr.TIME_SOURCE_INTERNAL
elif clock_source == X4xxClockMgr.CLOCK_SOURCE_EXTERNAL:
time_source = X4xxClockMgr.TIME_SOURCE_EXTERNAL
elif clock_source == X4xxClockMgr.CLOCK_SOURCE_GPSDO:
time_source = X4xxClockMgr.TIME_SOURCE_GPSDO
self.log.warning(
f"Time source '{old_time_source}' is an invalid selection with "
f"clock source '{clock_source}'. "
f"Coercing time source to '{time_source}'")
self.set_sync_source({
"clock_source": clock_source, "time_source": time_source})
def set_clock_source_out(self, enable):
"""
Allows routing the clock configured as source on the clk aux board to
the RefOut terminal. This only applies to internal, gpsdo and nsync.
"""
self._clk_mgr.set_clock_source_out(enable)
def get_time_sources(self):
" Returns list of valid time sources "
return self._clk_mgr.get_time_sources()
def set_time_source(self, time_source):
"""
Set a time source
This will call set_sync_source() internally, and use the current clock
source as a clock source. If the current clock source plus the requested
time source is not a valid combination, it will coerce the clock source
to a valid choice and print a warning.
"""
clock_source = self._clk_mgr.get_clock_source()
assert clock_source is not None
assert time_source is not None
if (clock_source, time_source) not in self._clk_mgr.valid_sync_sources:
old_clock_source = clock_source
if time_source == X4xxClockMgr.TIME_SOURCE_QSFP0:
clock_source = X4xxClockMgr.CLOCK_SOURCE_MBOARD
elif time_source == X4xxClockMgr.TIME_SOURCE_INTERNAL:
clock_source = X4xxClockMgr.CLOCK_SOURCE_MBOARD
elif time_source == X4xxClockMgr.TIME_SOURCE_EXTERNAL:
clock_source = X4xxClockMgr.CLOCK_SOURCE_EXTERNAL
elif time_source == X4xxClockMgr.TIME_SOURCE_GPSDO:
clock_source = X4xxClockMgr.CLOCK_SOURCE_GPSDO
self.log.warning(
'Clock source {} is an invalid selection with time source {}. '
'Coercing clock source to {}'
.format(old_clock_source, time_source, clock_source))
self.set_sync_source(
{"time_source": time_source, "clock_source": clock_source})
def set_sync_source(self, args):
"""
Selects reference clock and PPS sources. Unconditionally re-applies the
time source to ensure continuity between the reference clock and time
rates.
Note that if we change the source such that the time source is changed
to 'external', then we need to also disable exporting the reference
clock (RefOut and PPS-In are the same SMA connector).
"""
# Check the clock source, time source, and combined pair are valid:
clock_source = args.get('clock_source', self._clk_mgr.get_clock_source())
if clock_source not in self.get_clock_sources():
raise ValueError(f'Clock source {clock_source} is not a valid selection')
time_source = args.get('time_source', self._clk_mgr.get_time_source())
if time_source not in self.get_time_sources():
raise ValueError(f'Time source {time_source} is not a valid selection')
if (clock_source, time_source) not in self._clk_mgr.valid_sync_sources:
raise ValueError(
f'Clock and time source pair ({clock_source}, {time_source}) is '
'not a valid selection')
# Sanity checks complete. Now check if we need to disable the RefOut.
# Reminder: RefOut and PPSIn share an SMA. Besides, you can't export an
# external clock. We are thus not checking for time_source == 'external'
# because that's a subset of clock_source == 'external'.
# We also disable clock exports for 'mboard', because the mboard clock
# does not get routed back to the clocking aux board and thus can't be
# exported either.
if clock_source in (X4xxClockMgr.CLOCK_SOURCE_EXTERNAL,
X4xxClockMgr.CLOCK_SOURCE_MBOARD) and \
self._clocking_auxbrd:
self._clocking_auxbrd.export_clock(enable=False)
# Now the clock manager can do its thing.
ret_val = self._clk_mgr.set_sync_source(clock_source, time_source)
if ret_val == self._clk_mgr.SetSyncRetVal.NOP:
return
try:
# Re-set master clock rate. If this doesn't work, it will time out
# and throw an exception. We need to put the device back into a safe
# state in that case.
self.set_master_clock_rate(self._master_clock_rate)
# Restore the nco frequency to the same values as before the sync source
# was changed, to ensure the device transmission/acquisition continues at
# the requested frequency.
self.rfdc.rfdc_restore_nco_freq()
# Do the same for the calibration freeze state
self.rfdc.rfdc_restore_cal_freeze()
except RuntimeError as ex:
err = f"Setting clock_source={clock_source},time_source={time_source} " \
f"failed, falling back to {self._safe_sync_source}. Error: " \
f"{ex}"
self.log.error(err)
if args.get('__noretry__', False):
self.log.error("Giving up.")
else:
self.set_sync_source({**self._safe_sync_source, '__noretry__': True})
raise
def set_master_clock_rate(self, master_clock_rate):
"""
Sets the master clock rate by configuring the RFDC decimation and SPLL,
and then resetting downstream clocks.
"""
if master_clock_rate not in self.rfdc.master_to_sample_clk:
self.log.error('Unsupported master clock rate selection {}'
.format(master_clock_rate))
raise RuntimeError('Unsupported master clock rate selection')
sample_clock_freq, decimation, is_legacy_mode, halfband = \
self.rfdc.master_to_sample_clk[master_clock_rate]
for db_idx, _ in enumerate(self.dboards):
db_rfdc_resamp, db_halfband = self.rfdc.get_rfdc_resampling_factor(db_idx)
if db_rfdc_resamp != decimation or db_halfband != halfband:
msg = (f'master_clock_rate {master_clock_rate} is not compatible '
f'with FPGA which expected decimation {db_rfdc_resamp}')
self.log.error(msg)
raise RuntimeError(msg)
self.log.trace(f"Set master clock rate (SPLL) to: {master_clock_rate}")
self._clk_mgr.set_spll_rate(sample_clock_freq, is_legacy_mode)
self._master_clock_rate = master_clock_rate
self.rfdc.sync()
self._clk_mgr.config_pps_to_timekeeper(master_clock_rate)
def set_trigger_io(self, direction):
"""
Switch direction of clocking board Trigger I/O SMA socket.
IMPORTANT! Ensure downstream devices depending on TRIG I/O's output ignore
this signal when calling this method or re-run their synchronization routine
after calling this method. The output-enable control is async. to the output.
:param self:
:param direction: "off" trigger io socket unused
"pps_output" device will output PPS signal
"input" PPS is fed into the device from external
:return: success status as boolean
"""
OFF = "off"
INPUT = "input"
PPS_OUTPUT = "pps_output"
directions = [OFF, INPUT, PPS_OUTPUT]
if not self._clocking_auxbrd:
raise RuntimeError("No clocking aux board available")
if not direction in directions:
raise RuntimeError("Invalid trigger io direction (%s). Use one of %s"
% (direction, directions))
# Switching order of trigger I/O lines depends on requested direction.
# Always turn on new driver last so both drivers cannot be active
# simultaneously.
if direction == INPUT:
self.mboard_regs_control.set_trig_io_output(False)
self._clocking_auxbrd.set_trig(1, ClockingAuxBrdControl.DIRECTION_INPUT)
elif direction == PPS_OUTPUT:
self._clocking_auxbrd.set_trig(1, ClockingAuxBrdControl.DIRECTION_OUTPUT)
self.mboard_regs_control.set_trig_io_output(True)
else: # direction == OFF:
self.mboard_regs_control.set_trig_io_output(False)
self._clocking_auxbrd.set_trig(0)
return True
###########################################################################
# EEPROMs
###########################################################################
def get_db_eeprom(self, dboard_idx):
"""
See PeriphManagerBase.get_db_eeprom() for docs.
"""
try:
dboard = self.dboards[dboard_idx]
except IndexError:
error_msg = "Attempted to access invalid dboard index `{}' " \
"in get_db_eeprom()!".format(dboard_idx)
self.log.error(error_msg)
raise RuntimeError(error_msg)
db_eeprom_data = copy.copy(dboard.device_info)
return db_eeprom_data
###########################################################################
# Component updating
###########################################################################
# Note: Component updating functions defined by ZynqComponents
@no_rpc
def _update_fpga_type(self):
"""Update the fpga type stored in the updateable components"""
fpga_string = "{}_{}".format(
self.mboard_regs_control.get_fpga_type(),
self.rfdc.get_dsp_bw())
self.log.debug("Updating mboard FPGA type info to {}".format(fpga_string))
self.updateable_components['fpga']['type'] = fpga_string
#######################################################################
# Timekeeper API
#######################################################################
def get_master_clock_rate(self):
""" Return the master clock rate set during init """
return self._master_clock_rate
def get_clocks(self):
"""
Gets the RFNoC-related clocks present in the FPGA design
"""
# TODO: The 200 and 40 MHz clocks should not be hard coded, and ideally
# be linked to the FPGA image somehow
return [
{
'name': 'radio_clk',
'freq': str(self.get_master_clock_rate()),
'mutable': 'true'
},
{
'name': 'bus_clk',
'freq': str(200e6),
},
{
'name': 'ctrl_clk',
'freq': str(40e6),
}
]
###########################################################################
# GPIO API
###########################################################################
def get_gpio_banks(self):
"""
Returns a list of GPIO banks over which MPM has any control
"""
if self.dio_control is None:
return []
return self.dio_control.get_gpio_banks()
def get_gpio_srcs(self, bank: str):
"""
Return a list of valid GPIO sources for a given bank
"""
if self.dio_control is None:
return []
return self.dio_control.get_gpio_srcs(bank)
def get_gpio_src(self, bank: str):
"""
Return the currently selected GPIO source for a given bank. The return
value is a list of strings. The length of the vector is identical to
the number of controllable GPIO pins on this bank. CUSTOM is for
miscellaneous pin source, and USER_APP is for LabView pin source.
"""
if self.dio_control is None:
raise RuntimeError("Unable to query GPIO source: No valid DIO board installed.")
return self.dio_control.get_gpio_src(bank)
def set_gpio_src(self, bank: str, *src):
"""
Set the GPIO source for a given bank.
src input is big-endian
Usage:
> set_gpio_src <bank> <srcs>
> set_gpio_src GPIO0 PS DB1_RF1 PS PS MPM PS PS PS MPM USER_APP PS
"""
if self.dio_control is None:
raise RuntimeError("Unable to set GPIO source: No valid DIO board installed.")
self.dio_control.set_gpio_src(bank, *src)
###########################################################################
# Utility for validating RPU core number
###########################################################################
@no_rpc
def _validate_rpu_core_number(self, core_number):
if ((core_number < 0) or (core_number > 1)):
raise RuntimeError("RPU core number must be 0 or 1.")
###########################################################################
# Utility for validating RPU state change command
###########################################################################
@no_rpc
def _validate_rpu_state(self, new_state_command, previous_state):
if ((new_state_command != RPU_STATE_COMMAND_START)
and (new_state_command != RPU_STATE_COMMAND_STOP)):
raise RuntimeError("RPU state command must be start or stop.")
if ((new_state_command == RPU_STATE_COMMAND_START)
and (previous_state == RPU_STATE_RUNNING)):
raise RuntimeError("RPU already running.")
if ((new_state_command == RPU_STATE_COMMAND_STOP)
and (previous_state == RPU_STATE_OFFLINE)):
raise RuntimeError("RPU already offline.")
###########################################################################
# Utility for validating RPU firmware
###########################################################################
@no_rpc
def _validate_rpu_firmware(self, firmware):
file_path = path.join(RPU_REMOTEPROC_FIRMWARE_PATH, firmware)
if not path.isfile(file_path):
raise RuntimeError("Specified firmware does not exist.")
###########################################################################
# Utility for reading contents of a file
###########################################################################
@no_rpc
def _read_file(self, file_path):
self.log.trace("_read_file: file_path= %s", file_path)
with open(file_path, 'r') as f:
return f.read().strip()
###########################################################################
# Utility for writing contents of a file
###########################################################################
@no_rpc
def _write_file(self, file_path, data):
self.log.trace("_write_file: file_path= %s, data= %s", file_path, data)
with open(file_path, 'w') as f:
f.write(data)
###########################################################################
# RPU Image Deployment API
###########################################################################
def get_rpu_state(self, core_number, validate=True):
""" Report the state for the specified RPU core """
if validate:
self._validate_rpu_core_number(core_number)
return self._read_file(
path.join(
RPU_REMOTEPROC_PREFIX_PATH + str(core_number),
RPU_REMOTEPROC_PROPERTY_STATE))
def set_rpu_state(self, core_number, new_state_command, validate=True):
""" Set the specified state for the specified RPU core """
if not self._rpu_initialized:
self.log.warning(
"Failed to set RPU state: RPU peripheral not "\
"initialized.")
return RPU_FAILURE_REPORT
# OK, RPU is initialized, now go set its state:
if validate:
self._validate_rpu_core_number(core_number)
previous_state = self.get_rpu_state(core_number, False)
if validate:
self._validate_rpu_state(new_state_command, previous_state)
self._write_file(
path.join(
RPU_REMOTEPROC_PREFIX_PATH + str(core_number),
RPU_REMOTEPROC_PROPERTY_STATE),
new_state_command)
# Give RPU core time to change state (might load new fw)
poll_with_timeout(
lambda: previous_state != self.get_rpu_state(core_number, False),
RPU_MAX_STATE_CHANGE_TIME_IN_MS,
RPU_STATE_CHANGE_POLLING_INTERVAL_IN_MS)
# Quick validation of new state
resulting_state = self.get_rpu_state(core_number, False)
if ((new_state_command == RPU_STATE_COMMAND_START)
and (resulting_state != RPU_STATE_RUNNING)):
raise RuntimeError('Unable to start specified RPU core.')
if ((new_state_command == RPU_STATE_COMMAND_STOP)
and (resulting_state != RPU_STATE_OFFLINE)):
raise RuntimeError('Unable to stop specified RPU core.')
return RPU_SUCCESS_REPORT
def get_rpu_firmware(self, core_number):
""" Report the firmware for the specified RPU core """
self._validate_rpu_core_number(core_number)
return self._read_file(
path.join(
RPU_REMOTEPROC_PREFIX_PATH + str(core_number),
RPU_REMOTEPROC_PROPERTY_FIRMWARE))
def set_rpu_firmware(self, core_number, firmware, start=0):
""" Deploy the image at the specified path to the RPU """
self.log.trace("set_rpu_firmware")
self.log.trace(
"image path: %s, core_number: %d, start?: %d",
firmware,
core_number,
start)
if not self._rpu_initialized:
self.log.warning(
"Failed to deploy RPU image: "\
"RPU peripheral not initialized.")
return RPU_FAILURE_REPORT
# RPU is initialized, now go set firmware:
self._validate_rpu_core_number(core_number)
self._validate_rpu_firmware(firmware)
# Stop the core if necessary
if self.get_rpu_state(core_number, False) == RPU_STATE_RUNNING:
self.set_rpu_state(core_number, RPU_STATE_COMMAND_STOP, False)
# Set the new firmware path
self._write_file(
path.join(
RPU_REMOTEPROC_PREFIX_PATH + str(core_number),
RPU_REMOTEPROC_PROPERTY_FIRMWARE),
firmware)
# Start the core if requested
if start != 0:
self.set_rpu_state(core_number, RPU_STATE_COMMAND_START, False)
return RPU_SUCCESS_REPORT
#######################################################################
# Debugging
# Provides temporary methods for arbitrary hardware access while
# development for these components is ongoing.
#######################################################################
def peek_ctrlport(self, addr):
""" Peek the MPM Endpoint to ctrlport registers on the FPGA """
return '0x{:X}'.format(self.ctrlport_regs.peek32(addr))
def poke_ctrlport(self, addr, val):
""" Poke the MPM Endpoint to ctrlport registers on the FPGA """
self.ctrlport_regs.poke32(addr, val)
def peek_cpld(self, addr):
""" Peek the PS portion of the MB CPLD """
return '0x{:X}'.format(self.cpld_control.peek32(addr))
def poke_cpld(self, addr, val):
""" Poke the PS portion of the MB CPLD """
self.cpld_control.poke32(addr, val)
def peek_mb(self, addr):
""" Peek the MB Regs """
return '0x{:X}'.format(
self.mboard_regs_control.peek32(addr))
def poke_mb(self, addr, val):
""" Poke the MB CPLD """
self.mboard_regs_control.poke32(addr, val)
def peek_db(self, db_id, addr):
""" Peek the DB CPLD, even if the DB is not discovered by MPM """
assert db_id in (0, 1)
self.cpld_control.enable_daughterboard(db_id)
return '0x{:X}'.format(
self.ctrlport_regs.get_db_cpld_iface(db_id).peek32(addr))
def poke_db(self, db_id, addr, val):
""" Poke the DB CPLD, even if the DB is not discovered by MPM """
assert db_id in (0, 1)
self.cpld_control.enable_daughterboard(db_id)
self.ctrlport_regs.get_db_cpld_iface(db_id).poke32(addr, val)
def peek_clkaux(self, addr):
"""Peek the ClkAux DB over SPI"""
return '0x{:X}'.format(self._clocking_auxbrd.peek8(addr))
def poke_clkaux(self, addr, val):
"""Poke the ClkAux DB over SPI"""
self._clocking_auxbrd.poke8(addr, val)
###########################################################################
# Sensors
###########################################################################
def get_ref_lock_sensor(self):
"""
Return main refclock lock status. This is the lock status of the
reference and sample PLLs.
"""
lock_status = self._clk_mgr.get_ref_locked()
return {
'name': 'ref_locked',
'type': 'BOOLEAN',
'unit': 'locked' if lock_status else 'unlocked',
'value': str(lock_status).lower(),
}
def get_fpga_temp_sensor(self):
""" Get temperature sensor reading of the X4xx FPGA. """
self.log.trace("Reading FPGA temperature.")
return get_temp_sensor(["RFSoC"], log=self.log)
def get_main_power_temp_sensor(self):
"""
Get temperature sensor reading of PM-BUS devices which supply
0.85V power supply to RFSoC.
"""
self.log.trace("Reading PMBus Power Supply Chip(s) temperature.")
return get_temp_sensor(["PMBUS-0", "PMBUS-1"], log=self.log)
def get_scu_internal_temp_sensor(self):
""" Get temperature sensor reading of STM32 SCU's internal sensor. """
self.log.trace("Reading SCU internal temperature.")
return get_temp_sensor(["EC Internal"], log=self.log)
def get_internal_temp_sensor(self):
""" TODO: Determine how to interpret this function """
self.log.warning("Reading internal temperature is not yet implemented.")
return {
'name': 'temperature',
'type': 'REALNUM',
'unit': 'C',
'value': '-1'
}
def _get_fan_sensor(self, fan='fan0'):
""" Get fan speed. """
self.log.trace("Reading {} speed sensor.".format(fan))
fan_rpm = -1
try:
fan_rpm_all = ectool.get_fan_rpm()
fan_rpm = fan_rpm_all[fan]
except Exception as ex:
self.log.warning(
"Error occurred when getting {} speed value: {} "
.format(fan, str(ex)))
return {
'name': fan,
'type': 'INTEGER',
'unit': 'rpm',
'value': str(fan_rpm)
}
def get_fan0_sensor(self):
""" Get fan0 speed. """
return self._get_fan_sensor('fan0')
def get_fan1_sensor(self):
""" Get fan1 speed."""
return self._get_fan_sensor('fan1')
def get_gps_sensor_status(self):
"""
Get all status info of GPS in a single string. This is a debugging API.
"""
assert self._gps_mgr
self.log.trace("Reading all GPS status pins")
return f"""
{self.get_gps_lock_sensor()}
{self.get_gps_alarm_sensor()}
{self.get_gps_warmup_sensor()}
{self.get_gps_survey_sensor()}
{self.get_gps_phase_lock_sensor()}
"""
Reference in a new issue View git blame Copy permalink