This page documents internal details as well as the public interfaces.

Driver Utilities

General driver utilities

ska_low_cbf_fpga.driver._create_driver(driver_class, driver_args, logger)[source]

Create an ArgsFpgaDriver-derived class with given arguments.

Parameters:

driver_args (dict) – Arguments to pass to driver constructor.
driver_class (type[ArgsFpgaDriver]) – Derived class to create.
logger (Logger) – For logging.

Return type:

ArgsFpgaDriver

ska_low_cbf_fpga.driver._load_map(driver, firmware_path, fpgamap_path, logger)[source]

Create an ArgsMap (if possible), using the build timestamp from the driver.

Return type:: Optional[ArgsMap]

ska_low_cbf_fpga.driver._verify_registers(verify_registers, driver, map_, logger)[source]

Verify register values in FPGA.

Parameters:

verify_registers (Optional[dict[str, dict[str, int]]]) –

Register values to verify. e.g. To check for Correlator firmware:

{
    "system": {"firmware_personality": int.from_bytes(b"CORR", "big")}
}

Return type:

None

ska_low_cbf_fpga.driver.ami_fpga_present()[source]

Check if AMI-based FPGA hardware is available.

Return type:: bool
Returns:: True if AMI-based FPGA found, False otherwise
Raises:: RuntimeError – if ami_tool exists but doesn’t exit cleanly

ska_low_cbf_fpga.driver.create_driver_map_info(logger=None, mem_config='', fpgamap_path=None, firmware_path=None, device='0', partition=None, simulate=None, verify_registers=None, qdma_read=None, qdma_write=None)[source]

Create ArgsFpgaDriver, ArgsMap, FpgaHardwareInfo

Will use real FPGA hardware if present, unless simulate is explicitly set to True. If simulate is True or no FPGA is present, fpgamap_path must be provided for ArgsSimulator to use.

Parameters:

logger (Optional[Logger]) – Logger object to pass along for log outputs
mem_config (Union[list, str]) – FPGA memory access configuration
fpgamap_path (Optional[str]) – Path to fpgamap_<numbers>.py file to be used
firmware_path (Optional[str]) – path to a .pdi FPGA image (for AMI FPGA cards), or .xclbin FPGA kernel (for XRT FPGA cards). Optional for AMI FPGA cards (will use whatever is active on the card), required for XRT FPGA cards.
device (str) – PCIe Board:Device.Function address
partition (Optional[PartitionSpec]) – Partition number and boot device type to use for AMI FPGA cards.
simulate (Optional[bool]) – if True, simulate FPGA register/memory access
verify_registers (Optional[dict[str, dict[str, int]]]) – Register values used to verify that we have loaded the correct firmware image. Allows bypassing the long programming step for AMI cards without relying on UUIDs. e.g. {"system": {"commit_short_hash": 0x12345678} } This option may disappear if we can get the .pdi file UUID working. Not checked for XRT FPGAs - underlying driver relies on UUID.
qdma_read (Optional[str]) – Path to QDMA read (c2h) character device. AMI only.
qdma_write (Optional[str]) – Path to QDMA write (h2c) character device. AMI only.

Return type:

tuple[ArgsFpgaDriver, Optional[ArgsMap], Optional[FpgaHardwareInfo]]

Returns:

(driver, map, info)

driver is only created when either:

fpgamap_path is supplied or
an FPGA is present, firmware_path is supplied, and simulate is False.

To guarantee creation of driver, supply both fpgamap_path and a firmware file!

map is only created when an fpgamap is present (either explicitly specified or implicitly in the firmware directory). If no fpgamap provided or found, the return value of map will be None.

info is only created if XRT FPGA hardware is present. If no XRT FPGA, the return value of info will be None. simulate has no effect here, if info is not None then it is real hardware info.

Raises:

ValueError – if both types of firmware are given
RuntimeError – if supplied firmware not suitable for detected FPGA (AMI and XRT are mutually exclusive), unknown file extension in firmware_path, or mismatch between verify_registers and FPGA

ska_low_cbf_fpga.driver.xrt_fpga_present()[source]

Check if XRT-based FPGA hardware is available.

Return type:

bool

Returns:

True if XRT FPGA present, False if not

Raises:

FileNotFoundError – if pyxrt is available but xbutil is not
RuntimeError – if xbutil doesn’t exit cleanly

Describe Firmware

Tools for describing the FPGA firmware in use or available on a card.

ska_low_cbf_fpga.describe_firmware.describe_firmware(driver, map_dir='./', personality_to_package=<function <lambda>>)[source]

Describe the firmware in use by an FPGA driver.

Parameters:

driver (ArgsFpgaDriver) – An instance of ArgsFpgaDriver used to interact with the FPGA.
map_dir (str) – The directory to check for the ARGS FPGA register map file. Defaults to the current working directory. Note that subdirectories will be searched as well.
personality_to_package (Callable[[str], str]) – A callable that converts the FPGA personality four-letter code to the corresponding canonical package name. Defaults to a no-op lambda function that returns the input value unchanged.

Return type:

dict[str, UnionType[str, int, bool, None]]

Returns:

A dictionary containing entries:

”available”: A boolean indicating if the firmware is ready to use.
”map_build”: The ARGS map build timestamp, as a hexadecimal string. This value identifies the corresponding FPGA map file, which is required to access FPGA registers. If the driver fails to read the map build register, this value will be None (suggests an error in the underlying FPGA driver or hardware).
”package”: The package name for the firmware image, or None if unavailable (probably caused by a missing map file).
”version”: The firmware version or None if unavailable (probably caused by a missing map file).

ska_low_cbf_fpga.describe_firmware.describe_partitions(device, map_dir='./', personality_to_package=<function <lambda>>)[source]

Describe the firmware in each partition of a given AMI-based FPGA device.

This function inspects the partitions of an FPGA device, attempting to read information about the firmware in each partition. It compiles this data into a list of dictionaries, with one dictionary representing each partition’s details.

Warning

All partitions must be programmed with a valid ARGS firmware image. This function will boot from each partition to inspect it, and booting from an un-programmed flash partition can make the card unusable. A host reboot will be required to recover in this circumstance.

Parameters:

device (str) – The PCIe device identifier of the FPGA to inspect.
map_dir (str) – The directory containing ARGS FPGA register map files. Defaults to the current working directory.
personality_to_package (Callable[[str], str]) – A callable that converts the FPGA personality four-letter code to the corresponding canonical package name. Defaults to a no-op lambda function that returns the input value unchanged.

Return type:

list[dict[str, UnionType[str, int, bool, None]]]

Returns:

A list of dictionaries, each containing information about a partition on the device. Each dictionary contains entries:

”partition”: The partition index.

”available”: A boolean indicating if the firmware is ready to use.

”map_build”: The ARGS map build timestamp, as a hexadecimal string. This value identifies the corresponding FPGA map file, which is required to access FPGA registers. If the driver fails to read the map build register, this value will be None (suggests an error in the underlying FPGA driver or hardware).

”package”: The package name for the firmware image, or None if unavailable (probably caused by a missing map file).

”version”: The firmware version or None if unavailable (probably caused by a missing map file).

FPGA ICL

Instrument Control Layer

class ska_low_cbf_fpga.fpga_icl.FpgaPeripheral(driver, map_field_info, fpga_personality=None)[source]

Bases: FpgaUserInterface

__getattr__(name)[source]

Attribute access to fields

Return type:: IclFpgaField

__getitem__(item)[source]

Index style access to fields

Return type:: IclFpgaField

__init__(driver, map_field_info, fpga_personality=None)[source]

Base class that provides functions common to all FPGA Peripherals. Creates an instance of IclFpgaField for each field defined in the map. Populates _user_attributes if not set by derived class.

Parameters:

driver (ArgsFpgaDriver) – ArgsFpgaDriver instance, used to access FPGA
map_field_info (Dict[str, ArgsFieldInfo]) – Most likely a portion of an ArgsMap.
fpga_personality (Optional[NewType(FpgaPersonality, FpgaUserInterface)]) – personality applicable to this peripheral

__setattr__(key, value)[source]: Pass on attribute writes to the field, where appropriate

__setitem__(key, value)[source]: Pass on writes to the field

_field_config = {}

Configuration values for fields read from FPGA. e.g. to indicate to the control system that a field should be treated as an error condition:

_field_config = {"example": IclFpgaField(user_error=True)}

_not_user_attributes = {}: Attribute names to exclude from auto-discovery for the control system.

_not_user_methods = {}: Method names to exclude from auto-discovery for the control system.

_user_attributes = {}: Attributes to be exposed to the control system. Defaults to properties (if configured), or FPGA registers if no properties defined. Set to None if you really want no attributes exposed.

_user_methods = {}: Methods to be exposed to the control system. Set to None if you want no methods exposed.

property user_attributes: Set[str]: Attributes to be exposed to the control system.

property user_methods: Set[str]: Methods to be exposed to the control system.

class ska_low_cbf_fpga.fpga_icl.FpgaPersonality(driver, map_, hardware_info=None, logger=None)[source]

Bases: FpgaUserInterface

__getattr__(name)[source]: Access peripherals by attribute syntax

__getitem__(item)[source]

Index style access to peripheral objects

Return type:: FpgaPeripheral

__init__(driver, map_, hardware_info=None, logger=None)[source]

Base class that provides functions common to all FPGA Personalities. Creates FpgaPeripheral objects for each peripheral.

Parameters:

driver (ArgsFpgaDriver) – ArgsFpgaDriver instance, used to access FPGA
map – ARGS peripheral/register map.
hardware_info (Optional[FpgaHardwareInfo]) – Hardware monitoring interface (XrtInfo).
logger (Optional[Logger]) – Logger to report log messages to. Passed to peripherals.

_not_user_attributes = {}: Attribute names to exclude from auto-discovery for the control system.

_not_user_methods = {}: Method names to exclude from auto-discovery for the control system.

_user_attributes = {}: Attributes to be exposed to the control system. Defaults to properties (if configured), or FPGA registers if no properties defined. Set to None if you really want no attributes exposed.

_user_methods = {}: Methods to be exposed to the control system. Set to None if you want no methods exposed.

property default_interface: ArgsFpgaDriver: Temporary backwards-compatibility layer

property driver: ArgsFpgaDriver: Access to underlying ArgsFpgaDriver

property fw_personality: IclField[str]: Get the FPGA Firmware personality, decoded to a string

property fw_version: IclField[str]

Get the FPGA Firmware Version.

Returns:: version string formatted like the firmware packages: <major>.<minor>.<patch>[-<dev|main|pre|sim>.<commit hash>] “sim” indicates build type was zero (probably we are in simulation)

property peripherals: List[str]

Returns:: List of peripheral names

property read_memory: Callable: Access to read_memory function of underlying ArgsFpgaDriver

property user_attributes: Set[str]: Attributes to be exposed to the control system.

property user_methods: Set[str]: Methods to be exposed to the control system.

property write_memory: Callable: Access to write_memory function of underlying ArgsFpgaDriver

class ska_low_cbf_fpga.fpga_icl.FpgaUserInterface[source]

Bases: object

A common interface used by FpgaPeripheral and FpgaPersonality for exposing attributes and methods to the control system.

__init__()[source]

_not_user_attributes = {}: Attribute names to exclude from auto-discovery for the control system.

_not_user_methods = {}: Method names to exclude from auto-discovery for the control system.

_user_attributes = {}: Attributes to be exposed to the control system. Defaults to properties (if configured), or FPGA registers if no properties defined. Set to None if you really want no attributes exposed.

_user_methods = {}: Methods to be exposed to the control system. Set to None if you want no methods exposed.

property user_attributes: Set[str]: Attributes to be exposed to the control system.

property user_methods: Set[str]: Methods to be exposed to the control system.

ICL Fields

Instrument Control Layer (ICL) “Field” Objects.

These hold data (usually read from FPGA registers) and metadata.

class ska_low_cbf_fpga.icl_field.IclField(description=None, address=None, length=1, bit_offset=None, width=None, type_=~T, format='%i', user_write=True, user_error=False, value=None)[source]

Bases: ArgsFieldInfo, Generic[T]

ICL field, probably derived from one or more IclFpgaField s

__init__(description=None, address=None, length=1, bit_offset=None, width=None, type_=~T, format='%i', user_write=True, user_error=False, value=None)

user_error: bool = False: Should the control system treat a non-zero value as an error/alarm?

user_write: bool = True: Should the control system allow writing to this field? Note: not enforced at ICL!

class ska_low_cbf_fpga.icl_field.IclFpgaField(description=None, address=None, length=1, bit_offset=None, width=None, type_=<class 'ska_low_cbf_fpga.args_fpga.ArgsWordType'>, format='%i', user_write=True, user_error=False, value=<property object>, driver=None)[source]

Bases: IclField[ArgsWordType]

An ICL field that is linked to an ArgsFpgaDriver .

__getitem__(item)[source]

Array access for multi-word values

Return type:: ArgsWordType

__init__(description=None, address=None, length=1, bit_offset=None, width=None, type_=<class 'ska_low_cbf_fpga.args_fpga.ArgsWordType'>, format='%i', user_write=True, user_error=False, value=<property object>, driver=None)

__setitem__(key, value)[source]

Array index style access for setting part of a multi-word value. e.g. my_field[4] = 32 my_field[3:6] = np.ones(3, dtype=ArgsWordType)

If using a slice, value must be an array i.e. Don’t do this: my_field[3:4] = 5. Instead, use my_field[3] = 5 or even my_field[3:4] = np.array([5])

type_: alias of ArgsWordType

user_error: bool = False: Should the control system treat a non-zero value as an error/alarm?

user_write: bool = True: Should the control system allow writing to this field? Note: not enforced at ICL!

property value: int | ndarray | bool: Read current value from FPGA interface

ARGS FPGA

Base classes to interface with ARGS-based FPGA

class ska_low_cbf_fpga.args_fpga.ArgsFpgaDriver(logger=None, mem_config='', **kwargs)[source]

Bases: ArgsFpgaDriverIface

Base class for ARGS FPGA Drivers.

Extends interface class with startup steps common to all ARGS FPGAs.

__init__(logger=None, mem_config='', **kwargs)[source]

Parameters:

logger – a logging object with .debug, .info, etc functions
kwargs – optional extra arguments passed on to the derived class’s _setup function

get_map_build()[source]: Get the ARGS map build timestamp from the FPGA.

abstract read(source, length=None)

Read from FPGA register(s).

Parameters:

source – Start address (bytes)
length – Number of words to read

Return type:

ndarray | int

Returns:

int when length is 1, numpy array of ArgsWordType otherwise

abstract read_memory(index, size_bytes=None, offset_bytes=0)

Read a shared memory buffer.

Parameters:

size_bytes (Optional[int]) – number of bytes to transfer (transfers the whole buffer if not specified or None)
offset_bytes (int) – starting address
index (int) – Index of the shared buffer to save. Zero is the ARGS interchange buffer, which you probably don’t want.

Return type:

ndarray

Returns:

shared memory buffer

abstract write(destination, values)

Write to FPGA register(s).

Parameters:

destination – Start address (bytes)
values – One or more values to write to consecutive words

abstract write_memory(index, values, offset_bytes=0)

Write to a shared memory buffer.

Parameters:

index (int) – Index of the shared buffer to write to. Zero is the ARGS interchange buffer, which you probably don’t want.
values (ndarray) – Data to write.
offset_bytes (int) – Byte-based offset where values should start in buffer.

class ska_low_cbf_fpga.args_fpga.ArgsFpgaDriverIface[source]

Bases: ABC

Defines a pure-virtual Interface to a FPGA

abstract read(source, length=None)[source]

Read from FPGA register(s).

Parameters:

source – Start address (bytes)
length – Number of words to read

Return type:

ndarray | int

Returns:

int when length is 1, numpy array of ArgsWordType otherwise

abstract read_memory(index, size_bytes=None, offset_bytes=0)[source]

Read a shared memory buffer.

Parameters:

size_bytes (Optional[int]) – number of bytes to transfer (transfers the whole buffer if not specified or None)
offset_bytes (int) – starting address
index (int) – Index of the shared buffer to save. Zero is the ARGS interchange buffer, which you probably don’t want.

Return type:

ndarray

Returns:

shared memory buffer

abstract write(destination, values)[source]

Write to FPGA register(s).

Parameters:

destination – Start address (bytes)
values – One or more values to write to consecutive words

abstract write_memory(index, values, offset_bytes=0)[source]

Write to a shared memory buffer.

Parameters:

index (int) – Index of the shared buffer to write to. Zero is the ARGS interchange buffer, which you probably don’t want.
values (ndarray) – Data to write.
offset_bytes (int) – Byte-based offset where values should start in buffer.

exception ska_low_cbf_fpga.args_fpga.ArgsMagicError[source]

Bases: RuntimeError

Exception raised when FPGA does not report ARGS magic number.

__init__(*args, **kwargs)

with_traceback(): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class ska_low_cbf_fpga.args_fpga.ArgsWordType[source]

Bases: uint32

Data type used for ARGS words.

T

Scalar attribute identical to the corresponding array attribute.

Please see ndarray.T.

all()

Scalar method identical to the corresponding array attribute.

Please see ndarray.all.

any()

Scalar method identical to the corresponding array attribute.

Please see ndarray.any.

argmax()

Scalar method identical to the corresponding array attribute.

Please see ndarray.argmax.

argmin()

Scalar method identical to the corresponding array attribute.

Please see ndarray.argmin.

argsort()

Scalar method identical to the corresponding array attribute.

Please see ndarray.argsort.

astype()

Scalar method identical to the corresponding array attribute.

Please see ndarray.astype.

base

Scalar attribute identical to the corresponding array attribute.

Please see ndarray.base.

bit_count() → int

Computes the number of 1-bits in the absolute value of the input. Analogous to the builtin int.bit_count or popcount in C++.

Examples

>>> np.uint32(127).bit_count()
7

byteswap()

Scalar method identical to the corresponding array attribute.

Please see ndarray.byteswap.

choose()

Scalar method identical to the corresponding array attribute.

Please see ndarray.choose.

clip()

Scalar method identical to the corresponding array attribute.

Please see ndarray.clip.

compress()

Scalar method identical to the corresponding array attribute.

Please see ndarray.compress.

conjugate()

Scalar method identical to the corresponding array attribute.

Please see ndarray.conjugate.

copy()

Scalar method identical to the corresponding array attribute.

Please see ndarray.copy.

cumprod()

Scalar method identical to the corresponding array attribute.

Please see ndarray.cumprod.

cumsum()

Scalar method identical to the corresponding array attribute.

Please see ndarray.cumsum.

data: Pointer to start of data.

denominator: denominator of value (1)

diagonal()

Scalar method identical to the corresponding array attribute.

Please see ndarray.diagonal.

dtype: Get array data-descriptor.

dump()

Scalar method identical to the corresponding array attribute.

Please see ndarray.dump.

dumps()

Scalar method identical to the corresponding array attribute.

Please see ndarray.dumps.

fill()

Scalar method identical to the corresponding array attribute.

Please see ndarray.fill.

flags: The integer value of flags.

flat: A 1-D view of the scalar.

flatten()

Scalar method identical to the corresponding array attribute.

Please see ndarray.flatten.

getfield()

Scalar method identical to the corresponding array attribute.

Please see ndarray.getfield.

imag: The imaginary part of the scalar.

is_integer() → bool

Return True if the number is finite with integral value.

Added in version 1.22.

Examples

>>> np.int64(-2).is_integer()
True
>>> np.uint32(5).is_integer()
True

item()

Scalar method identical to the corresponding array attribute.

Please see ndarray.item.

itemset()

Scalar method identical to the corresponding array attribute.

Please see ndarray.itemset.

itemsize: The length of one element in bytes.

max()

Scalar method identical to the corresponding array attribute.

Please see ndarray.max.

mean()

Scalar method identical to the corresponding array attribute.

Please see ndarray.mean.

min()

Scalar method identical to the corresponding array attribute.

Please see ndarray.min.

nbytes: The length of the scalar in bytes.

ndim: The number of array dimensions.

newbyteorder(new_order='S', /)

Return a new dtype with a different byte order.

Changes are also made in all fields and sub-arrays of the data type.

The new_order code can be any from the following:

‘S’ - swap dtype from current to opposite endian
{‘<’, ‘little’} - little endian
{‘>’, ‘big’} - big endian
{‘=’, ‘native’} - native order
{‘|’, ‘I’} - ignore (no change to byte order)

Parameters

new_orderstr, optional: Byte order to force; a value from the byte order specifications above. The default value (‘S’) results in swapping the current byte order.

Returns

new_dtypedtype: New dtype object with the given change to the byte order.

nonzero()

Scalar method identical to the corresponding array attribute.

Please see ndarray.nonzero.

numerator: numerator of value (the value itself)

prod()

Scalar method identical to the corresponding array attribute.

Please see ndarray.prod.

ptp()

Scalar method identical to the corresponding array attribute.

Please see ndarray.ptp.

put()

Scalar method identical to the corresponding array attribute.

Please see ndarray.put.

ravel()

Scalar method identical to the corresponding array attribute.

Please see ndarray.ravel.

real: The real part of the scalar.

repeat()

Scalar method identical to the corresponding array attribute.

Please see ndarray.repeat.

reshape()

Scalar method identical to the corresponding array attribute.

Please see ndarray.reshape.

resize()

Scalar method identical to the corresponding array attribute.

Please see ndarray.resize.

round()

Scalar method identical to the corresponding array attribute.

Please see ndarray.round.

searchsorted()

Scalar method identical to the corresponding array attribute.

Please see ndarray.searchsorted.

setfield()

Scalar method identical to the corresponding array attribute.

Please see ndarray.setfield.

setflags()

Scalar method identical to the corresponding array attribute.

Please see ndarray.setflags.

shape: Tuple of array dimensions.

size: The number of elements in the gentype.

sort()

Scalar method identical to the corresponding array attribute.

Please see ndarray.sort.

squeeze()

Scalar method identical to the corresponding array attribute.

Please see ndarray.squeeze.

std()

Scalar method identical to the corresponding array attribute.

Please see ndarray.std.

strides: Tuple of bytes steps in each dimension.

sum()

Scalar method identical to the corresponding array attribute.

Please see ndarray.sum.

swapaxes()

Scalar method identical to the corresponding array attribute.

Please see ndarray.swapaxes.

take()

Scalar method identical to the corresponding array attribute.

Please see ndarray.take.

tofile()

Scalar method identical to the corresponding array attribute.

Please see ndarray.tofile.

tolist()

Scalar method identical to the corresponding array attribute.

Please see ndarray.tolist.

tostring()

Scalar method identical to the corresponding array attribute.

Please see ndarray.tostring.

trace()

Scalar method identical to the corresponding array attribute.

Please see ndarray.trace.

transpose()

Scalar method identical to the corresponding array attribute.

Please see ndarray.transpose.

var()

Scalar method identical to the corresponding array attribute.

Please see ndarray.var.

view()

Scalar method identical to the corresponding array attribute.

Please see ndarray.view.

ska_low_cbf_fpga.args_fpga.EXCHANGE_BUF_CONFIG = MemConfig(size=131072, shared=True): ARGS register interchange buffer configuration.

exception ska_low_cbf_fpga.args_fpga.FpgaRegisterError[source]

Bases: RuntimeError

Exception raised when FPGA register read fails.

__init__(*args, **kwargs)

with_traceback(): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class ska_low_cbf_fpga.args_fpga.MemConfig(size, shared)[source]

Bases: object

Memory buffer configuration info

__init__(size, shared)

ska_low_cbf_fpga.args_fpga.WORD_SIZE = 4: Size of ARGS words, in bytes.

ska_low_cbf_fpga.args_fpga.mem_config_check(mem_config='')[source]

Parameters:: mem_config (Union[list, str]) – a list of MemConfig tuples (<size in bytes>, <shared?>) the first list item is used to send/receive register values

ska_low_cbf_fpga.args_fpga.mem_parse(memory_config='')[source]

Convert a string-formatted list of memory configurations to a list of MemConfigs.

The register interchange buffer is automatically included as the first element. Do not list the register buffer in the input string!

Parameters:: memory_config (str) – a colon-separated string <size><unit><s|i> size: int unit: k, M, G (powers of 1024) s: shared i: FPGA internal e.g. ‘128Ms:1Gi’
Return type:: List[MemConfig]
Returns:: list of MemConfigs, decoded from the string

ska_low_cbf_fpga.args_fpga.str_from_int_bytes(n_bytes, precision=2)[source]

Automatically scale a number of bytes for printing.

No decimals are shown for exact conversions, e.g. 1024 -> “1 KiB”

Parameters:

n_bytes (int) – Number of bytes
precision (int) – decimal places to show

Return type:

str

Returns:

A number and units, e.g. ‘1.5 KiB’

ARGS Map

class ska_low_cbf_fpga.args_map.AddressInfo(name, length)[source]

Minimal set of information about an FPGA register address. (a collection of these is used for address number -> name lookups)

__init__(name, length)

class ska_low_cbf_fpga.args_map.ArgsFieldInfo(description=None, address=None, length=1, bit_offset=None, width=None)[source]

__init__(description=None, address=None, length=1, bit_offset=None, width=None)

class ska_low_cbf_fpga.args_map.ArgsMap(spec)[source]

ARGS FPGA map file decoder.

__init__(spec)[source]

Parameters:: spec (dict) – ARGS register map

__setitem__(key, value)[source]

Should only be called internally, during object creation.

Parameters:

key – peripheral name
value – dict of ArgsFieldInfo

Raises:

NotImplementedError – if called after ArgsMap init

_decode_map(fpga_map)[source]: Given an FPGAMAP nested dictionary structure, extract relevant info for its RAM and Register fields.

classmethod create_from_file(build, map_dir)[source]

Parameters:

build (int) – ARGS map build timestamp
map_dir (str) – directory to find ARGS map file in

ska_low_cbf_fpga.args_map.all_register_addresses(fpga_map)[source]

Creates a lookup table that can be used to convert byte based addresses to full register names.

Parameters:: fpga_map (dict) – FPGAMAP object
Return type:: Dict[int, AddressInfo]
Returns:: e.g. {40: "fpga.system.system.time_uptime"}

ska_low_cbf_fpga.args_map.get_name_and_offset(registers, address)[source]

Get register name and offset into register for a given address. Looks up the nearest lesser address and finds the relative offset. Checks that this is within the defined register length.

Parameters:

registers (Dict[int, AddressInfo]) – use all_register_addresses() to create this
address (int) – address to look up

Return type:

(int, int)

Returns:

(name, word offset) - e.g. ("array.my_array.data", 4)

Raises:

ValueError – if not a valid address

ska_low_cbf_fpga.args_map.load_fpgamap_from_file(map_file_path)[source]

Load the FPGAMAP variable from a specified python file.

Parameters:: map_file_path (str) – path to .py file containing FPGAMAP dict
Return type:: dict

ska_low_cbf_fpga.args_map.load_map(build, map_dir)[source]

Load FPGA register map file from a directory.

Looks for a file named with the build date of the running firmware image, recursively searching the given directory.

Parameters:

build (int) – Hexadecimal representation of this is used to select filename, fpgamap_<build>.py
map_dir (str) – Directory containing the map file. No other paths are searched!

ARGS AMI

ARGS FPGA driver using ami_tool.

class ska_low_cbf_fpga.args_ami_tool.ArgsAmi(logger=None, mem_config='', **kwargs)[source]

Bases: ArgsFpgaDriver

FPGA driver that wraps ami_tool.

__init__(logger=None, mem_config='', **kwargs)

Parameters:

logger – a logging object with .debug, .info, etc functions
kwargs – optional extra arguments passed on to the derived class’s _setup function

_check_magic(): Check that FPGA contains the correct ‘Magic Number’.

_get_bar_address(raw_address)[source]

Get PCIe BAR number and byte address from a given raw byte address.

Return type:: tuple[int, int]

_init_buffers()[source]: Initialise memory buffers.

_load_firmware()[source]: Load a firmware file into the FPGA.

_setup(*, device='0', partition=None, pdi_file=None, bars=(2, 4), bar_address_split=18446744073709551615, qdma_read=None, qdma_write=None, hbm_offsets=(), **kwargs)[source]

Set up the FPGA driver.

Parameters:

device (str) – PCIe BDF address. Can be partial, don’t need the zeros.
boot_type – The boot device type.
partition (Optional[PartitionSpec]) – Flash memory partition to activate and boot from. If pdi_file is given, it will be loaded to this partition unless the image in the partition has a matching UUID.
pdi_file (Optional[str]) – Path to .pdi file to load. If not given, use whatever is already on the card.
bars (Optional[tuple[int, ...]]) – PCIe BAR numbers to use, defaults to (2, 4).
bar_address_split (int) – Byte address of first register in second PCIe BAR. The default value is the largest possible (64 bit maximum unsigned int) so we can read ARGS magic number and map build before loading the map. Then, using the map, a user can find the address of the real split value. This driver does not automatically read the true split value, as we have no knowledge of the ARGS register address map.
qdma_read (Optional[str]) – QDMA read (c2h) character device path, used for HBM access.
qdma_write (Optional[str]) – QDMA write (h2c) character device path, used for HBM access.
hbm_offsets (tuple[int, ...]) – HBM address offsets. One value per HBM buffer, in order.

get_map_build(): Get the ARGS map build timestamp from the FPGA.

property n_partitions: int: Number of Flash Partition Table (FPT) entries.

read(source, length=1)[source]

Read FPGA registers.

Return type:: Union[ndarray, int]

read_memory(index, size_bytes=None, offset_bytes=0)[source]

Read from HBM.

Note: Unlike ArgsXrt, we do not care if the memory buffer is marked as ‘shared’ in the memory configuration string.

Parameters:

size_bytes (Optional[int]) – Number of bytes to transfer. Should be a multiple of WORD_SIZE, will be rounded down to a multiple of WORD_SIZE. (transfers the whole buffer if not specified or None)
offset_bytes (int) – Starting address.
index (int) – Index of the HBM buffer to read. For compatibility with ArgsXrt, the index of the first HBM buffer is 1.

Return type:

ndarray

Returns:

HBM contents, as ArgsWordType

write(destination, values)[source]

Write to FPGA registers.

Return type:: None

write_memory(index, values, offset_bytes=0)[source]

Write to HBM.

Note: Unlike ArgsXrt, we do not care if the memory buffer is marked as ‘shared’ in the memory configuration string.

Parameters:

index (int) – Index of the HBM buffer to write to. For compatibility with ArgsXrt, the index of the first HBM buffer is 1.
values (ndarray) – Data to write.
offset_bytes (int) – Starting address.

class ska_low_cbf_fpga.args_ami_tool.BootDeviceType(value)[source]

Bases: Enum

AMI boot device types.

ska_low_cbf_fpga.args_ami_tool.DMA_MAX_SIZE = 268435456: Maximum number of bytes per DMA transaction. Not sure exactly what the limiting factor is, possibly QDMA configuration options.

class ska_low_cbf_fpga.args_ami_tool.PartitionSpec(boot_type, partition)[source]

Bases: object

AMI Partition Specifier.

__init__(boot_type, partition)

ska_low_cbf_fpga.args_ami_tool._parse_read_string(read, n_words)[source]

Decode the output from an ami_tool bar_rd command.

Parameters:

read (str) – output of ami_tool bar_rd command.
n_words (int) – number of words that were read.

Return type:

ndarray

ska_low_cbf_fpga.args_ami_tool._uuid_mismatch(cfgmem_program_output)[source]

Check if incoming and current UUIDs differ.

Parameters:: cfgmem_program_output (str) – output of ami_tool cfgmem_program command
Return type:: bool
Returns:: True if current and incoming UUIDs differ, False otherwise

ska_low_cbf_fpga.args_ami_tool.parse_hbm_addresses_file(filename)[source]

Read HBM address offsets from a file.

Parameters:: filename (Path | str) – Path to an HBM address map file. The File must contain one integer value per line, which is used as the QDMA address offset for the respective HBM buffer.
Return type:: tuple[int, ...]

AMI Info

class ska_low_cbf_fpga.ami_info.AmiInfo(hwmon_dir, sys_dir_0)[source]

Bases: FpgaHardwareInfo

Hardware monitoring for AMI devices via sysfs.

__init__(hwmon_dir, sys_dir_0)[source]

_read_hw(filename)[source]: Read text from a file in our hwmon directory.

_read_sys(filename)[source]: Read text from a file in our sysfs directory.

property fpga_power: IclField[float]: Get FPGA power consumption in Watts.

property fpga_temperature: IclField[int]: Get FPGA temperature in degrees Celsius.

property hbm_temperature: IclField[int]: Get HBM temperature in degrees Celsius.

property mac_addresses: IclField[str]

Get Ethernet MAC addresses.

Returns:: Array of str, formatted like “00:01:02:03:04:05”.

property pcie_12v_current: IclField[float]: Get PCIe 12V power rail’s current.

property pcie_12v_voltage: IclField[float]: Get PCIe 12V power rail voltage reading.

property power_supply_12v_current: IclField[float]: Get 12V AUX total current (sum of two).

property power_supply_12v_voltage: IclField[float]: Get 12V AUX voltage (one of the two that deviates furthest from 12V).

ska_low_cbf_fpga.ami_info._int_from_mac_str(mac)[source]

Convert MAC address str to int.

Parameters:: mac – MAC address string that may contain seperator characters e.g. “00:01:02:03:04:05”
Return type:: int

ARGS XRT

class ska_low_cbf_fpga.args_xrt.ArgsXrt(logger=None, mem_config='', **kwargs)[source]

Bases: ArgsFpgaDriver

pyxrt-based FPGA Driver. Requires a .xclbin file.

MINUS_ONE_16BIT = 65535: -1 as a 16 bit signed integer (0xFFFF)

__init__(logger=None, mem_config='', **kwargs)

Parameters:

logger – a logging object with .debug, .info, etc functions
kwargs – optional extra arguments passed on to the derived class’s _setup function

_check_magic(): Check that FPGA contains the correct ‘Magic Number’.

_init_buffers()[source]: Create XRT & Numpy buffers.

_load_firmware()[source]: Load the binary to the FPGA (if required) and initialise our kernel object.

_read_args_page(address_start, n_bytes=None)[source]

Read one block of register values into host memory buffer.

Parameters:

address_start (int) – Byte address to start reading from
n_bytes (Optional[int]) – Number of bytes to read

_setup(xcl_file, mem_config='', device='0', **kwargs)[source]

Parameters:

xcl_file (str) – path to a .xclbin FPGA kernel
mem_config (Union[list, str]) – a list of MemConfig tuples (<size in bytes>, <shared?>) the first list item is used to send/receive register values
device (str) – PCIe Board:Device.Function address

_write_args_page(destination, values)[source]

Write one block of register values from host buffer to FPGA.

Parameters:

destination – Byte address to start writing to
values – Array of values to be written to consecutive words

get_map_build(): Get the ARGS map build timestamp from the FPGA.

read(source, length=1)[source]

Read values from FPGA.

Parameters:

source (int) – Byte address to start reading from
length (int) – Number of words to read

Returns:

value(s) from FPGA

read_memory(index, size_bytes=None, offset_bytes=0)[source]

Read a shared memory buffer.

Parameters:

size_bytes (Optional[int]) – number of bytes to transfer (transfers the whole buffer if not specified or None)
offset_bytes (int) – starting address
index (int) – Index of the shared buffer to save. Zero is the ARGS interchange buffer, which you probably don’t want.

Return type:

ndarray

Returns:

shared memory buffer

write(destination, values)[source]

Write values to FPGA.

Parameters:

destination (int) – FPGA byte address where writes should start
values (Union[int, ndarray]) – value(s) to write, if more than one value they must be words, to be written to consecutive words (i.e. byte addresses increment by WORD_SIZE)

write_memory(index, values, offset_bytes=0)[source]

Write to a shared memory buffer.

Parameters:

index (int) – Index of the shared buffer to write to. Zero is the ARGS interchange buffer, which you probably don’t want.
values (ndarray) – Data to write.
offset_bytes (int) – Byte-based offset where values should start in buffer.

ska_low_cbf_fpga.args_xrt.XRT_TIMEOUT = 5: Timeout for FPGA read/write actions (milliseconds)

ska_low_cbf_fpga.args_xrt._wait_for_completion(kernel_task, timeout)[source]

Wait for FPGA register transaction to complete.

Similar purpose to pyxrt.run.wait() except we throw an exception for errors or timeouts rather than failing silently.

TODO: If timeouts do happen sometimes, should they be handled gracefully e.g retry?

Parameters:

kernel_task (run) – pyxrt kernel run object to monitor
timeout (int) – timeout in milliseconds

Raises:

RuntimeError – if the command does not succeed in a timely manner

XRT Info

class ska_low_cbf_fpga.xrt_info.XrtInfo(device)[source]

Bases: FpgaHardwareInfo

Hardware info monitoring via pyxrt xrt_info_device.

Access via item index, e.g. my_xrt_info["thermal"]. Some flattening of data structures is performed.

_INFO_PARAMS = {'bdf': <class 'str'>, 'dynamic_regions': 'json', 'electrical': 'json', 'host': 'json', 'interface_uuid': <class 'str'>, 'kdma': <class 'bool'>, 'm2m': <class 'bool'>, 'max_clock_frequency_mhz': <class 'int'>, 'mechanical': 'json', 'memory': 'json', 'name': <class 'str'>, 'nodma': <class 'bool'>, 'offline': <class 'bool'>, 'pcie_info': 'json', 'platform': 'json', 'thermal': 'json'}: Mapping from known hardware info item keys to their data types.

__getitem__(item)[source]

Access info parameters via item index syntax.

Parameters:: item (str) – probably one of the values defined in _INFO_PARAMS

__init__(device)[source]

__weakref__: list of weak references to the object (if defined)

property fpga_power: IclField[float]: Get FPGA power consumption in Watts.

property fpga_temperature: IclField[int]: Get FPGA temperature in degrees Celsius.

property hbm_temperature: IclField[int]: Get HBM temperature in degrees Celsius.

property mac_addresses: IclField[str]

Get Ethernet MAC addresses.

Returns:: Array of str, formatted like “00:01:02:03:04:05”.

property pcie_12v_current: IclField[float]: Get PCIe 12V power rail’s current.

property pcie_12v_voltage: IclField[float]: Get PCIe 12V power rail voltage reading.

property power_supply_12v_current: IclField[float]: Get 12V power rail’s current.

property power_supply_12v_voltage: IclField[float]: Get power rail 12 volt reading.

property xclbin_uuid: str: Get the UUID of the active xclbin firmware.

Hardware Info

Hardware monitoring interface

class ska_low_cbf_fpga.hardware_info.FpgaHardwareInfo[source]

Hardware monitoring interface definition.

The properties defined here must be implemented in driver-specific derived classes.

__weakref__: list of weak references to the object (if defined)

abstract property fpga_power: IclField[float]: Get FPGA power consumption in Watts.

abstract property fpga_temperature: IclField[int]: Get FPGA temperature in degrees Celsius.

abstract property hbm_temperature: IclField[int]: Get HBM temperature in degrees Celsius.

abstract property mac_addresses: IclField[str]

Get Ethernet MAC addresses.

Returns:: Array of str, formatted like “00:01:02:03:04:05”.

abstract property pcie_12v_current: IclField[float]: Get PCIe 12V power rail’s current.

abstract property pcie_12v_voltage: IclField[float]: Get PCIe 12V power rail voltage reading.

abstract property power_supply_12v_current: IclField[float]: Get 12V power rail’s current.

abstract property power_supply_12v_voltage: IclField[float]: Get power rail 12 volt reading.

Logging

Logging configuration & translation functions

class ska_low_cbf_fpga.log.ConversionFormat(value)[source]: Log conversion output formats.

class ska_low_cbf_fpga.log.LogEntry(time, mode, name, offset, values, length, address, level, source)[source]

Details of one log entry.

__init__(time, mode, name, offset, values, length, address, level, source)

ska_low_cbf_fpga.log._write_log_entry(conversion_format, entry, file_out, hex_vals)[source]: Write one log entry in the requested format.

ska_low_cbf_fpga.log.convert_log()[source]: Command-Line Interface - convert a log file.

ska_low_cbf_fpga.log.convert_log_contents(log, register_lookup, conversion_format, file_out, include_reads=False, include_others=False, hex_vals=False, filter_spec=None)[source]

Convert the contents of a log file.

Parameters:

log (TextIO) – input log file
register_lookup (Dict[int, AddressInfo]) – lookup table
conversion_format (ConversionFormat) – choice of output format
file_out (TextIO) – output file to write to
include_reads (bool) – include FPGA reads in output?
include_others (bool) – include non-FPGA register log lines in output?
hex_vals (bool) – convert values to hex? (only relevant for HUMAN mode)

filter_spec (Union[dict, TextIO, None]) –

filtering configuration as dict or JSON in file. e.g.

{
  "select_words": {
      # word address range to keep (inclusive)
      "spead_sdp.spead_params.data": [0, 7167],
      "spead_sdp_2.spead_params.data": [0, 7167],
  },
  "ignore": [
      "spead_sdp.spead_ctrl.trigger_packet",
      "spead_sdp_2.spead_ctrl.trigger_packet",
  ],
  "keep_only_last": True,
}

Return type:

None

ska_low_cbf_fpga.log.log_to_file(logger, filename, level=10, hex_vals=False)[source]

Add a file handler to a logger, with our predefined log format, and configure numpy to ensure full arrays are printed on a single line. Warning: numpy print options are a global setting!

Parameters:

logger (Logger) – Logger to configure
filename (str) – File name to write (append) to
level (int) – Logging level, defaults to DEBUG to capture driver read/write
hex_vals (bool) – convert values to hexadecimal?

Return type:

None

ska_low_cbf_fpga.log.stop_log_to_file(logger)[source]

Stop register logging.

Parameters:: logger (Logger) – Logger to configure
Return type:: None