Drivers API

Drivers provide low-level hardware control and protocol implementations.

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: Driver

Mount and copy files to a USB mass storage device.

Supports both local-only (MassStorageDevice) and remote-exporter (NetworkUSBMassStorage / USBMassStorage) bindings. When bound to a remote resource, pmount/pumount/mkdir run on the exporter host via ssh (labgrid’s command_prefix) and file copies use scp.

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'}}

sw_downloads_template = 'https://swdownloads.analog.com/cse/boot_partition_files/{release}/latest_boot_partition.tar.gz'

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, get_boot_files=False)

Download the specified Kuiper release version if not already cached. :param release_version: Version of the Kuiper release to download. If None, uses the version from kuiper_resource. :type release_version: str

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

FPGA/JTAG Drivers

XilinxJTAGDriver

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

Bases: 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

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

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