Drivers API

Drivers provide low-level hardware control and protocol implementations.

On this page:

• Power Drivers — VesyncPowerDriver, CyberPowerDriver, CyberPowerPdu, HomeAssistantPowerDriver, HomeAssistantClient

• Shell and File Transfer — ADIShellDriver

• Storage Drivers — MassStorageDriver

• Kuiper Drivers — KuiperDLDriver

• Cloudsmith Drivers — CloudsmithDLDriver

• FPGA/JTAG Drivers — XilinxJTAGDriver

• Network Drivers — TFTPServerDriver, Utility Classes

• Software Installer — SoftwareInstallerDriver

Power Drivers

VesyncPowerDriver

class adi_lg_plugins.drivers.vesyncdriver.VesyncPowerDriver(target, name)

Bases: Driver, PowerResetMixin, PowerProtocol

VesyncPowerDriver - Driver using a Vesync Smart Outlet to control a target’s power - https://github.com/webdjoe/pyvesync. Uses pyvesync tool to control the outlet.

bindings: Dict[str, Any] = {'vesync_outlet': {'VesyncOutlet'}}

on()

Turn on all configured VeSync outlets.

This method powers on all outlets specified in the VesyncOutlet resource configuration. If multiple outlets are configured, they will all be turned on sequentially.

Raises:

Exception – If outlet control fails or outlets are not found.

off()

Turn off all configured VeSync outlets.

This method powers off all outlets specified in the VesyncOutlet resource configuration. If multiple outlets are configured, they will all be turned off sequentially.

Raises:

Exception – If outlet control fails or outlets are not found.

reset()

Perform a power reset cycle on all outlets.

This method turns off the outlets, waits for the configured delay period, then turns them back on. This is useful for hard-resetting hardware.

The delay duration is configured in the VesyncOutlet resource.

Raises:

Exception – If outlet control fails.

cycle()

Power cycle all outlets (same as reset).

Alias for reset(). Turns off the outlets, waits for the configured delay, then turns them back on.

Raises:

Exception – If outlet control fails.

get()

Get the current power state of all outlets.

Returns:

True if all configured outlets are on, False otherwise.

Return type:

bool

__init__(target, name)

Method generated by attrs for class VesyncPowerDriver.

Return type:

None

CyberPowerDriver

class adi_lg_plugins.drivers.cyberpowerdriver.CyberPowerDriver(target, name)

Bases: Driver, PowerResetMixin, PowerProtocol

CyberPowerDriver - Driver using a CyberPower PDU to control a target’s power

__init__(target, name)

Method generated by attrs for class CyberPowerDriver.

Return type:

None

bindings: Dict[str, Any] = {'cyberpower_outlet': {'CyberPowerOutlet'}}

on()

Turn on the configured CyberPower PDU outlet.

Uses SNMP to send an ‘immediateOn’ command to the outlet specified in the CyberPowerOutlet resource configuration.

Raises:

CyberPowerPduException – If SNMP communication fails.

off()

Turn off the configured CyberPower PDU outlet.

Uses SNMP to send an ‘immediateOff’ command to the outlet specified in the CyberPowerOutlet resource configuration.

Raises:

CyberPowerPduException – If SNMP communication fails.

reset()

Perform a power reset cycle on the outlet.

This method turns off the outlet, waits for the configured delay period, then turns it back on. Useful for hard-resetting hardware.

The delay duration is configured in the CyberPowerOutlet resource.

Raises:

CyberPowerPduException – If SNMP communication fails.

cycle()

Power cycle the outlet (same as reset).

Alias for reset(). Turns off the outlet, waits for the configured delay, then turns it back on.

Raises:

CyberPowerPduException – If SNMP communication fails.

CyberPowerPdu

class adi_lg_plugins.drivers.cyberpowerdriver.CyberPowerPdu(host)

Bases: object

Class to query & control a CyberPower PDU via SNMP.

Tested on the PDU15SWHVIEC8FNET. I don’t understand SNMP well enough to have any idea if this would be expected to work on other models.

This class is basically just a piece of copy-pasted pysnmp code and a depository for comments.

Parameters:

host (str) – IP address or hostname of the PDU on the network

outlet_state_oids = {'cancelPendingCommand': 7, 'delayedOff': 5, 'delayedOn': 4, 'delayedReboot': 6, 'immediateOff': 2, 'immediateOn': 1, 'immediateReboot': 3, 'outletIdentify': 8}

__init__(host)

async async_set_outlet_on(outlet, on)

Set an outlet on or off (async version for pysnmp >= 7.0.0)

Parameters:

• outlet – Which outlet to set the power for (for my model this is in the range 1 through 8)

• on – INVALID ATM True means turn it on, False means turn it off

set_outlet_on(outlet, on)

Set an outlet on or off (synchronous wrapper)

Parameters:

• outlet – Which outlet to set the power for (for my model this is in the range 1 through 8)

• on – INVALID ATM True means turn it on, False means turn it off

HomeAssistantPowerDriver

class adi_lg_plugins.drivers.homeassistantdriver.HomeAssistantPowerDriver(target, name)

Bases: Driver, PowerResetMixin, PowerProtocol

HomeAssistantPowerDriver - Driver using a Home Assistant switch/outlet to control a target’s power via the Home Assistant REST API.

bindings: Dict[str, Any] = {'ha_outlet': {'HomeAssistantOutlet'}}

on()

Turn on the configured Home Assistant switch.

off()

Turn off the configured Home Assistant switch.

reset()

Power reset: off, delay, on.

cycle()

Power cycle (same as reset).

get()

Get the current power state.

Returns:

True if the switch is on, False otherwise.

Return type:

bool

__init__(target, name)

Method generated by attrs for class HomeAssistantPowerDriver.

Return type:

None

HomeAssistantClient

class adi_lg_plugins.drivers.homeassistantdriver.HomeAssistantClient(url, token)

Bases: object

Client for Home Assistant REST API switch control.

Parameters:

• url – Base URL of the Home Assistant instance

• token – Long-lived access token

__init__(url, token)

turn_on(entity_id)

Turn on an entity.

turn_off(entity_id)

Turn off an entity.

get_state(entity_id)

Get the current state of an entity.

Returns:

True if the entity state is ‘on’, False otherwise.

Return type:

bool

Shell and File Transfer

ADIShellDriver

class adi_lg_plugins.drivers.shelldriver.ADIShellDriver(target, name, prompt, login_prompt, username, password=None, keyfile='', login_timeout=60, console_ready='', await_login_timeout=2, post_login_settle_time=0)

Bases: CommandMixin, Driver, CommandProtocol, FileTransferProtocol

ADIShellDriver - Driver to execute commands on the shell ADIShellDriver binds on top of a ConsoleProtocol.

On activation, the ADIShellDriver will look for the login prompt on the console, and login to provide shell access.

Parameters:

• prompt (regex) – the shell prompt to detect

• login_prompt (regex) – the login prompt to detect

• username (str) – username to login with

• password (str) – password to login with

• keyfile (str) – keyfile to bind mount over users authorized keys

• login_timeout (int) – optional, timeout for login prompt detection

• console_ready (regex) – optional, pattern used by the kernel to inform the user that a console can be activated by pressing enter.

• await_login_timeout (int) – optional, time in seconds of silence that needs to pass before sending a newline to device.

• post_login_settle_time (int) – optional, seconds of silence after logging in before check for a prompt. Useful when the console is interleaved with boot output which may interrupt prompt detection.

bindings: Dict[str, Any] = {'console': <class 'labgrid.protocol.consoleprotocol.ConsoleProtocol'>}

on_activate()

Called by the Target when this object has been activated

on_deactivate()

Called by the Target when this object has been deactivated

run(cmd, timeout=30.0, codec='utf-8', decodeerrors='strict')

Run a command

get_status()

Returns the status of the shell-driver. 0 means not connected/found, 1 means shell

put_ssh_key(keyfile_path)

put_bytes(buf, remotefile)

Upload a file to the target. Will silently overwrite the remote file if it already exists.

Parameters:

• buf (bytes) – file contents

• remotefile (str) – destination filename on the target

Raises:

ExecutionError – if something went wrong

put(localfile, remotefile)

Upload a file to the target. Will silently overwrite the remote file if it already exists.

Parameters:

• localfile (str) – source filename on the local machine

• remotefile (str) – destination filename on the target

Raises:

• IOError – if the provided localfile could not be found

• ExecutionError – if something else went wrong

get_bytes(remotefile)

Download a file from the target.

Parameters:

remotefile (str) – source filename on the target

Returns:

(bytes) file contents

Raises:

ExecutionError – if something went wrong

get(remotefile, localfile)

Download a file from the target. Will silently overwrite the local file if it already exists.

Parameters:

• remotefile (str) – source filename on the target

• localfile (str) – destination filename on the local machine (can be relative)

Raises:

• IOError – if localfile could not be written

• ExecutionError – if something went wrong

run_script(data, timeout=60)

Upload a script to the target and run it.

Parameters:

• data (bytes) – script data

• timeout (int) – timeout for the script to finish execution

Returns:

str, stderr: str, return_value: int)

Return type:

Tuple of (stdout

Raises:

ExecutionError – if something went wrong

run_script_file(scriptfile, *args, timeout=60)

Upload a script file to the target and run it.

Parameters:

• scriptfile (str) – source file on the local file system to upload to the target

• *args – (list of str): any arguments for the script as positional arguments

• timeout (int) – timeout for the script to finish execution

Returns:

str, stderr: str, return_value: int)

Return type:

Tuple of (stdout

Raises:

• ExecutionError – if something went wrong

• IOError – if the provided localfile could not be found

get_default_interface_device_name(version=4)

Retrieve the default route’s interface device name.

Parameters:

version (int) – IP version

Returns:

Name of the device of the default route

Raises:

ExecutionError – if no or multiple routes are set up

get_ip_addresses(device=None)

Retrieves IP addresses for given interface name.

Note that although the return type is named IPv4Interface/IPv6Interface, it contains an IP address with the corresponding network prefix.

Parameters:

device (str) – Name of the interface to query, defaults to default route’s device name.

Returns:

List of IPv4Interface or IPv6Interface objects

run_uboot(cmd, timeout=30)

Runs the specified command on the shell and returns the output.

Parameters:

• cmd (str) – command to run on the shell

• timeout (int) – optional, how long to wait for completion

Returns:

if successful, None otherwise

Return type:

Tuple[List[str],List[str], int]

get_status_uboot()

Retrieve status of the UBootDriver. 0 means inactive, 1 means active.

Returns:

status of the driver

Return type:

int

__init__(target, name, prompt, login_prompt, username, password=None, keyfile='', login_timeout=60, console_ready='', await_login_timeout=2, post_login_settle_time=0)

Method generated by attrs for class ADIShellDriver.

Return type:

None

Storage Drivers

MassStorageDriver

class adi_lg_plugins.drivers.massstoragedriver.MassStorageDriver(target, name, partition=None, mount_label='lg_mass_storage')

Bases: RemoteExecMixin, Driver

Mount and copy files to a USB mass storage device.

Supports both local-only (test runner == exporter) and remote-exporter bindings. When the bound resource is proxied from a coordinator, pmount/pumount/mkdir run on the exporter host and file copies are staged there over a single reused ssh connection (see RemoteExecMixin).

Specify partition when the bound resource points at a whole block device (e.g. /dev/sdb) rather than a specific partition; its value is the absolute partition path on the exporter host — a raw device (/dev/sdb1) or a stable symlink (/dev/disk/by-partuuid/…).

bindings: Dict[str, Any] = {'mass_storage': {'MassStorageDevice'}}

mount_partition()

Mount the configured partition at /media/<mount_label>.

unmount_partition()

Unmount the mass storage device partition.

copy_file(src, dst)

Copy a local file onto the mass storage device.

Parameters:

• src – source file path on the test runner host.

• dst – destination path relative to the mount point.

update_files()

Batch-copy files listed in mass_storage.file_updates (local-only path mapping).

__init__(target, name, partition=None, mount_label='lg_mass_storage')

Method generated by attrs for class MassStorageDriver.

Return type:

None

Kuiper Drivers

KuiperDLDriver

class adi_lg_plugins.drivers.kuiperdldriver.KuiperDLDriver(target, name)

Bases: Driver

KuiperDLDriver - Driver to download and manage Kuiper releases and provide files to the target device.

bindings: Dict[str, Any] = {'kuiper_resource': {'KuiperRelease'}}

cache_datafile = 'cache_info.json'

check_cached(release_version=None)

Check if the specified Kuiper release version is cached locally. :param release_version: Version of the Kuiper release to check. If None, uses the version from kuiper_resource. :type release_version: str

Returns:

True if the release is cached, False otherwise.

Return type:

bool

download_release(release_version=None)

Download the specified Kuiper release version if not already cached.

get_full_image_path(release_version=None)

Return the cached full SD image (.img) path for the configured release.

Downloads + extracts the release first if it isn’t cached. Caller is responsible for activating/deactivating the driver.

get_boot_files_from_release(get_all_files=False)

add_files_to_target(filename)

Add a file to the target device.

Parameters:

filename (str) – Path to the file to add to the target.

__init__(target, name)

Method generated by attrs for class KuiperDLDriver.

Return type:

None

Cloudsmith Drivers

CloudsmithDLDriver

class adi_lg_plugins.drivers.cloudsmithdldriver.CloudsmithDLDriver(target, name)

Bases: Driver

Driver to resolve and download Cloudsmith boot artifacts.

Exposes the same get_boot_files_from_release() / _boot_files contract as KuiperDLDriver so it drops into the FPGA SoC boot strategies.

bindings: Dict[str, Any] = {'cloudsmith_resource': {'CloudsmithRelease'}}

cache_datafile = 'cache_info.json'

check_cached(version)

Return the cached boot-file path for version if present, else None.

download_release(version=None)

Resolve, download, and cache the boot artifact; return its local path.

get_boot_file_path(version=None)

Ensure the artifact is downloaded and return its local path.

get_boot_files_from_release()

Strategy-facing contract: populate and return _boot_files.

__init__(target, name)

Method generated by attrs for class CloudsmithDLDriver.

Return type:

None

FPGA/JTAG Drivers

XilinxJTAGDriver

class adi_lg_plugins.drivers.xilinxjtagdriver.XilinxJTAGDriver(target, name)

Bases: RemoteExecMixin, Driver

Program Xilinx FPGAs via JTAG using xsdb.

Bindings:

xilinxdevicejtag: XilinxDeviceJTAG resource (JTAG target IDs + bitstream/kernel paths

as seen by the host that runs xsdb).

xilinxvivado: XilinxVivadoTool resource (vivado_path / xsdb_path).

bindings: Dict[str, Any] = {'xilinxdevicejtag': {'XilinxDeviceJTAG'}, 'xilinxvivado': {'XilinxVivadoTool'}}

connect_jtag()

Connect to JTAG interface.

flash_bitstream()

Flash the FPGA bitstream via JTAG.

download_kernel()

Download Linux kernel image to Microblaze processor.

start_execution()

Start kernel execution on Microblaze processor.

load_bitstream_and_kernel_and_start()

Load bitstream + kernel, then run the Microblaze.

disconnect_jtag()

Disconnect from JTAG interface.

load_zynq_uboot(ps7_init_tcl, uboot_elf, a9_target_name='*Cortex-A9 MPCore #0', bitstream_path=None, fsbl_elf=None)

JTAG-bootstrap U-Boot on a Zynq-7000 device.

The board can be in any boot state — xsdb will rst -system first to clear residual DDR/PS state before sourcing the board-specific ps7_init.tcl. Used for SD-card recovery when BootROM cannot load FSBL from a corrupted card.

The a9_target_name filter is used instead of an integer target index because Zynq-7000 xsdb target ordering shifts when the PL is loaded; the name-pattern form matches Xilinx’s generated wrappers and is stable across Vivado versions.

Parameters:

• ps7_init_tcl (str)

• uboot_elf (str)

• a9_target_name (str)

• bitstream_path (str | None)

• fsbl_elf (str | None)

Return type:

None

load_and_run_elf(elf_path, a9_target_name='*Cortex-A9 MPCore #0', bitstream_path=None, ps7_init_tcl=None)

JTAG-load and start an arbitrary bare-metal ELF (e.g. no-os firmware).

Generalizes load_zynq_uboot() to any ELF that runs directly on a Zynq core (no FSBL/U-Boot chain). The same xsdb sequence is used: connect → rst -system → [fpga] → [ps7_init] → dow elf → con. The optional bitstream_path programs the PL first (required when the firmware touches FPGA-fabric peripherals), and ps7_init_tcl runs the board PS init — both are produced by the no-os build’s HDL .xsa.

Paths are resolved to absolute before being embedded in the xsdb TCL: xsdb runs the script from its own working directory (not the caller’s), so a relative dow/fpga -f path would fail to open.

Parameters:

• elf_path (str)

• a9_target_name (str)

• bitstream_path (str | None)

• ps7_init_tcl (str | None)

Return type:

None

stop_zynq_cpu(a9_target_name='*Cortex-A9 MPCore #0')

Halt the A9 #0 core — used between failed bootstrap attempts.

Parameters:

a9_target_name (str)

Return type:

None

__init__(target, name)

Method generated by attrs for class XilinxJTAGDriver.

Return type:

None

Network Drivers

TFTPServerDriver

class adi_lg_plugins.drivers.tftpserverdriver.TFTPServerDriver(target, name)

Bases: Driver

TFTPServerDriver provides a pure Python TFTP server.

bindings: Dict[str, Any] = {'resource': <class 'adi_lg_plugins.resources.tftpserver.TFTPServerResource'>}

on_activate()

Called by the Target when this object has been activated

on_deactivate()

Called by the Target when this object has been deactivated

get_server_address()

__init__(target, name)

Method generated by attrs for class TFTPServerDriver.

Return type:

None

Utility Classes

class adi_lg_plugins.drivers.kuiperdldriver.Downloader

Utility class for downloading and verifying Kuiper Linux releases.

This class handles downloading release archives from ADI’s servers, verifying MD5 checksums, and extracting compressed archives. It supports both .xz and .zip compressed formats and displays progress bars using tqdm.

The Downloader is used internally by KuiperDLDriver but can also be used standalone via the kuiperdl CLI tool.

Example

>>> dl = Downloader()
>>> rel = dl.releases("2023_R2_P1")
>>> dl.download(rel["link"], rel["zipname"])
>>> dl.check(rel["zipname"], rel["zipmd5"])
>>> dl.extract(rel["zipname"], rel["imgname"])

releases(release='2019_R1')

retry_session(retries=3, backoff_factor=0.3, status_forcelist=(429, 500, 502, 504), session=None)

download(url, fname)

check(fname, ref, find_img=False)

extract(inname, outname)

extract_xz(inname, outname)

extract_zip(inname, outdir)

class adi_lg_plugins.drivers.imageextractor.IMGFileExtractor(img_path, logger=None)

Extract files from disk image (.img) files using pytsk3.

This utility class provides methods to inspect partitions, list files, and extract individual files or directories from disk image files without mounting them. It’s primarily used by KuiperDLDriver to extract boot files from Kuiper Linux release images.

Parameters:

• img_path (str) – Path to the disk image file.

• logger (Logger, optional) – Logger instance for debug output. If None, prints to stdout.

Example

>>> extractor = IMGFileExtractor("kuiper.img")
>>> partitions = extractor.get_partitions()
>>> fs = extractor.open_filesystem(partitions[0]["start"])
>>> extractor.extract_file(fs, "/Image", "./output/Image")
>>> extractor.close()

__init__(img_path, logger=None)

log(message)

get_partitions()

List all partitions in the IMG file

open_filesystem(partition_offset)

Open filesystem at a specific partition offset

list_files(fs, path='/')

Recursively list all files in a directory

extract_file(fs, file_path, output_path)

Extract a single file

extract_directory(fs, source_path, output_dir)

Extract an entire directory recursively

close()

Close the IMG file handle

class adi_lg_plugins.drivers.cloudsmithdldriver.Downloader

Streaming downloader with retry + sha256 verification.

Copied from KuiperDLDriver’s Downloader (kept independent). Only the pieces needed for individual-file downloads are retained: a retrying session and a streaming download that returns the sha256 digest.

retry_session(retries=3, backoff_factor=0.3, status_forcelist=(429, 500, 502, 504), session=None)

download(url, fname, headers=None)

Stream url to fname and return the sha256 hexdigest.

adi_lg_plugins.drivers.cloudsmithdldriver.get_latest_bootfiles(owner, repo, fpga_carrier=None, daughter_card=None, vfilter=None, vnot=None, filename=None, token=None, pin=None)

Resolve the matching Cloudsmith package, returning the raw package dict.

Without pin the newest matching package (by uploaded_at) is returned; with pin the package whose version equals pin is returned. The returned dict carries cdn_url, checksums version and uploaded_at plus a parsed _info entry.

adi_lg_plugins.drivers.cloudsmithdldriver.parse_version_info(version_str, tags, repo=None)

Parse a Cloudsmith version string into structured metadata.

Returns None (rather than raising) for unparseable strings, ignored systems, or unknown carriers, so one unexpected package never crashes resolution.

Software Installer

SoftwareInstallerDriver

class adi_lg_plugins.drivers.softwareinstaller.SoftwareInstallerDriver(target, name)

Bases: Driver

SoftwareInstallerDriver - Driver to install software, clone repos, copy directories, and run builds/tests on a DUT.

bindings: Dict[str, Any] = {'command': <class 'labgrid.protocol.commandprotocol.CommandProtocol'>, 'file_transfer': <class 'labgrid.protocol.filetransferprotocol.FileTransferProtocol'>}

install_package(package_name, update=False)

Installs a package using the detected package manager.

Parameters:

• package_name (str) – Name of the package to install.

• update (bool) – Whether to update package lists before installing.

clone_repo(repo_url, destination, branch=None)

Clones a git repository to the destination.

Parameters:

• repo_url (str) – URL of the git repository.

• destination (str) – Remote path to clone into.

• branch (str) – Optional branch or tag to checkout.

copy_directory(local_path, remote_path)

Copies a local directory to the remote path.

Parameters:

• local_path (str) – Local directory path.

• remote_path (str) – Remote directory path (parent directory must exist or be created).

run_build(command, directory)

Runs a build command in a specific directory.

Parameters:

• command (str) – Build command (e.g., ‘make’, ‘cargo build’).

• directory (str) – Directory to run the build in.

run_binary(binary_path, args='', directory=None)

Runs a binary on the target.

Parameters:

• binary_path (str) – Path to the binary.

• args (str) – Arguments for the binary.

• directory (str) – Working directory.

run_test(test_command, directory=None)

Runs a test command.

Parameters:

• test_command (str) – Command to run tests.

• directory (str) – Working directory.

__init__(target, name)

Method generated by attrs for class SoftwareInstallerDriver.

Return type:

None