Quick-start

The bare minimum example: creating a driver, loading an ARGS map, and supplying them to an FpgaPersonality.

xcl = "/path/to/firmware.xclbin"
driver, args_map = create_driver_and_map(xcl_file=xcl)
fpga = FpgaPersonality(driver, args_map)

# read
x = int(fpga.peripheral_name.field_name)
z = fpga.another_peripheral.field.value
my_slice = fpga.peripheral_name.multi_word_field[2:4]

# write
fpga.peripheral_name.field_name = y
fpga.peripheral_name.multi_word_field[2:4] = np.zeros(2, dtype=ArgsWordType)

Creating a Personality

The main thing is to provide the mapping from peripheral names (as reported in the map) to Python classes, using the class variable _peripheral_class.

from ska_low_cbf_fpga import FpgaPersonality
from packetiser import Packetiser

class ExampleFpga(FpgaPersonality):
    _peripheral_class = {
        "packetiser": Packetiser,
    }

Creating a Peripheral

  • Derive your class from FpgaPeripheral.

  • Write peripheral-specific methods & properties. You can use attribute style access to register values, or use the underlying peripheral dict-style access. Return an IclField so the next layer up (control system mapping) has all the info it needs.

    • Use user_write and user_error fields in the object returned to provide configuration information to the control system.

  • Specify methods & attributes that should be exposed to the control system using _user_methods, _not_user_methods, _user_attributes, and _not_user_attributes.

  • By default, all public methods (those named without a leading underscore) will be included.

  • Special keys can be included in _user_attributes to control automatic discovery of user attributes. By default, discovery is performed only if _user_attributes is empty, and registers are only discovered if no properties are discovered. * DISCOVER_PROPERTIES: Always discover properties * DISCOVER_REGISTERS: Always discover registers * DISCOVER_ALL: Always discover both properties & registers

class ExamplePeripheral(FpgaPeripheral):
    # Hints for the control system

    # If not specified, _user_methods will include all public methods.
    # If you only want to expose certain methods, list them explicitly:
    # _user_methods = {"on"}
    # If you want to expose most with a few exclusions, use _not_user_methods:
    _not_user_methods = {"internal_use"}
    # Use _user_methods = None if you don't want to expose any methods.

    # If not specified, _user_attributes will include all properties.
    # If no properties exist, it will include all fields.
    # Set _user_attributes = None if you do not want to expose any fields/properties.
    # Manual specification example:
    # _user_attributes = {"reset_active", "something_wrong", "dst_ip"}
    # Use the DISCOVER_ constants to discover more than the defaults.
    _user_attributes = {"counter", DISCOVER_PROPERTIES}
    # Use _not_user_attributes to exclude things from discovery:
    _not_user_attributes = {"secret_property"}

    # Simple read-only property example
    @property
    def reset_active(self) -> IclField[bool]:
        # attribute-style access (beware when overloading names)
        reset_active = self.module_reset == 1
        return IclField(description="Reset Active", type_=bool, value=reset_active, format="%s", user_write=False)

    # Error condition example
    @property
    def something_wrong(self) -> IclField[bool]:
        error = self.error_1 or self.error_2
        return IclField(description="something wrong", type_=bool, value=error, user_error=True, user_write=False)

    # Method examples
    def on(self):
        # dict-style access
        self["control_vector"] = 3

    def internal_use(self, setting):
        self["my_register"] = setting

    # Get/Set property example
    DATA_IP_DST = 8  # class variable for address within RAM buffer

    @property
    def dst_ip(self) -> IclField[str]:
        """Destination IPv4 address"""
        dst_ip = self.data[DATA_IP_DST]
        ip = socket.inet_ntoa(int(dst_ip).to_bytes(4, "big"))
        return IclField(description="Destination IPv4 Address", type_=str, value=ip, format="%s", user_write=True)

    @dst_ip.setter
    def dst_ip(self, ip: str):
        """Set destination IPv4 address"""
        self.data[DATA_IP_DST] = socket.inet_aton(ip)

    # For sake of example for _not_user_attributes
    @property
    def secret_property(self) -> IclField[int]:
        secret = (self.counter // 123) * 456
        return IclField(description="Number for use in our code only", type_=int, value=secret)

Advanced Peripheral Options

Special field configuration can be provided in _field_config, for example if you wish to override the ARGS-provided description, specify a non-integer type, etc.

The keys in this dictionary must match the name of the field from the ARGS map.

class FieldsExample(FpgaPeripheral):
    _field_config = {
        "description_example": IclFpgaField(description="my desc."),
        "error_example": IclFpgaField(user_error=True),
        "read_only_suggestion_to_control_system": IclFpgaField(user_write=False),
    }

Status Monitoring

Implemented via pyxrt.

If Alveo hardware is present, the create_driver_map_info function (implemented in driver.py) will create an instance of an XrtInfo object, which can be accessed through FpgaPersonality.info. Most information is accessed via array item syntax, with an exception of xclbin_uuid as a property. Some parameters return complex data structures.

Warning: when no Alveo hardware is present, .info will be None. Classes deriving from FpgaPersonality must allow for this.

# check if hardware monitoring is available
# (downstream user to implement)
if fpga.info is None:
    print("No hardware monitoring available")
    return

# get UUID of xclbin file loaded to card (str)
fpga.info.xclbin_uuid

# get card's serial number (str)
fpga.info["platform"]["controller"]["card_mgmt_controller"]["serial_number"]

# Ethernet MAC address
fpga.info["platform"]["macs"][0]["address"]

Note that the data types of info items vary:

for item in fpga.info:
    print(item, type(fpga.info[item]))
bdf <class 'str'>
dynamic_regions <class 'str'>
electrical <class 'dict'>
host <class 'dict'>
interface_uuid <class 'str'>
kdma <class 'bool'>
m2m <class 'bool'>
max_clock_frequency_mhz <class 'int'>
mechanical <class 'dict'>
memory <class 'dict'>
name <class 'str'>
nodma <class 'bool'>
offline <class 'bool'>
pcie_info <class 'dict'>
platform <class 'dict'>
thermal <class 'list'>

Logging

For FPGA debugging purposes, logging functions are provided to record all register transactions.

A helper function log_to_file is provided to configure logging to a file with a standard format. In the near future, a log parser will be provided to convert logs to a form suitable to load into an FPGA.

You can configure register logging like this, noting that some details will depend on your implementation details including other loggers that you have configured and whether you wish to log register transactions via them.

import logging
import sys
import numpy as np
from ska_low_cbf_fpga.log import log_to_file

# this works for the fpga CLI, but may differ in control system implementations
root_logger = self.logger

# Optional - find existing handlers, prevent them from logging register read/writes
for handler in root_logger.handlers:
    handler.setLevel(logging.INFO)

log_to_file(fpga.driver.logger, "my_log_file.txt", hex_vals=True)

The log file will contain entries like this:

2023-06-20 09:11:38,590|ArgsSimulator|DEBUG|Read address 0x6c, length 1: [0x0]
2023-06-20 09:11:38,590|ArgsSimulator|DEBUG|Write address 0x6c: [0x1]
2023-06-20 09:11:40,233|ArgsSimulator|DEBUG|Read address 0x6c, length 1: [0x1]

Log Conversion

The convert_log utility can be used to convert register logs to different formats:

  • Human-Readable

  • Testbench

  • FPGA Registers (text file for loading back to FPGA)

Logs can optionally be filtered to facilitate simulation or debugging. Filtering is configured using a JSON file.

For example, to drop SPEAD data words 7168+, ignore SPEAD packet triggers, and keep only the last value of all registers:

{
  "select_words": {
    "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
}