saymrwulf/uhd
1
0
Fork 0
mirror of https://github.com/saymrwulf/uhd.git synced 2026-05-14 20:58:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
uhd/mpm/python/usrp_mpm/periph_manager/x4xx.py
Martin Braun cf04329058 mpm: x4xx: Introduce X4xxClockPolicy 
This new class encapsulates all clocking-related infos into a single
file and makes it easier to modify clocking settings.
2023-05-12 15:43:29 -05:00

1104 lines
44 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.compat_num import CompatNumber
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 make_mb_cpld_ctrl
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.periph_manager.x4xx_clock_policy import get_clock_policy
from usrp_mpm.dboard_manager.x4xx_db_iface import X4xxDboardIface
X400_FPGA_COMPAT = (7, 9)
# The compat number at which remote streaming was added:
X400_REMOTE_STREAMING_COMPAT = (7, 9)
X400_DEFAULT_ENABLE_PPS_EXPORT = True
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"
@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._gps_mgr = None
self.clk_mgr = None
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):
"""
Read the FPGA compat number and throw an exception if the compat numbers
don't match up (i.e., this version of MPM requires a newer FPGA compat
number).
If the numbers match, we check the FPGA compat number for available
features.
"""
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
)
if CompatNumber(actual_compat) >= CompatNumber(X400_REMOTE_STREAMING_COMPAT):
self.fpga_features.add('remote_udp_streaming')
self.log.debug(
"FPGA supports the following features: {}"
.format(", ".join(self.fpga_features)))
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.
Reminder: This is the first function that is run that actually talks to
hardware *after* PeriphManagerBase.__init__() has been executed. The
only bits of hardware that have been initialized/read out are the
motherboard and daughterboard EEPROMs.
After this, we initialize the daughterboards, so we have to make sure
everything else, in particular the clocks, are running as desired.
"""
# 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()
self._add_public_methods(self._clocking_auxbrd, "clkaux")
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
# Init CPLD before talking to clocking ICs
cpld_spi_node = dt_symbol_get_spidev('mb_cpld')
# This factory function will check signature and compat-rev, and
# therefore could throw if the CPLD is not compatible.
self.cpld_control = make_mb_cpld_ctrl(cpld_spi_node, self.log)
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__(), we don't know yet which
# MCR will be requested by the user. 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 later (in init()).
self.clk_mgr = X4xxClockMgr(
args,
clk_policy=get_clock_policy(self.mboard_info, self.dboard_infos, args, self.log),
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'),
allow_overwrite=True
)
# Overlay must be applied after clocks have been configured. This will
# load the FPGA bitfile.
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()
# The clock manager needs access to this object -- we will make that
# available after the RFDC object is created.
# Create control for RFDC
self.rfdc = X4xxRfdcCtrl(self.log)
self._add_public_methods(
self.rfdc, prefix="",
filter_cb=lambda name, method: not hasattr(method, '_norpc')
)
self._update_fpga_type()
# Now that we have applied the overlay, we can query the FPGA
# capabilities. This will allow us to figure out a good default MCR,
# and we set that. Remember that earlier, when we turned on the clocks,
# we just picked any valid clock rate so we would be able to communicate
# with all the devices. After this call, all clocks will be set to useful
# and valid values.
self.clk_mgr.finalize_init(args, self.mboard_regs_control, self.rfdc)
# 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
args = self._update_default_args(args)
# 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)
# Initialize all clocking-related stuff: Master clock rate, time/clock
# source, initialize all the PLLs, ...
self.clk_mgr.init(args)
# 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)
)
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
self.clk_mgr.deinit()
super(x4xx, self).deinit()
self.ctrlport_regs.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.tear_down()
# 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)
###########################################################################
# 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. Note: Most of the API calls are defined in
# X4xxClockManager.
###########################################################################
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
###########################################################################
# 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