driverclient 0.2.0

Driver Server client with an embeddable connector for config injection at runtime.

driverclient

Embeddable client for a Driver Server local repo. Scans a Windows machine's hardware, resolves and installs drivers from the repo, and captures drivers back up to it. Designed to be driven programmatically from a host application (for example a PyQt front-end) via a small connector class.

Install

pip install driverclient

Usage

Create one DriverClient per process and call .run():

from driverclient import DriverClient

client = DriverClient(
    local_repo_url="http://REPO_HOST:8000",
    node_key="YOUR_NODE_KEY",
)
result = client.run("capture-all")   # or client.run() for the configured default

Configuration

Config can be supplied three ways and is merged with this priority (highest wins):

DEFAULTS  <  DS_CLIENT_CONFIG env file  <  config_path file  <  keyword args

• Args only — pass keys directly:

  DriverClient(local_repo_url="http://x:1", node_key="k")
  

• File only — pass a JSON file used as the base:

  DriverClient(config_path="/etc/driverclient.json")
  

• Both — file as the base, individual keys overridden by kwargs:

  DriverClient(config_path="/etc/driverclient.json", node_key="override")
  

Passing nothing preserves the default behavior (env file → packaged config.json → built-in defaults).

A dummy config.json ships inside the package as a template. Replace REPLACE_WITH_YOUR_NODE_KEY and point local_repo_url at your repo — or, better, supply real values at runtime through the connector.

Commands

scan, resolve, resolve-and-install, capture-all, capture-missing, wu-update, wu-full, automate.

client.run() with no argument uses DS_CLIENT_COMMAND, then the default_command from config.

Progress events

Every operation reports progress in real time through a structured event stream (added in 0.2.0). Pass an on_event callback to run() and it is invoked synchronously for each step:

def on_event(ev):        # ev is a driverclient.ClientEvent
    print(ev.phase, ev.status, ev.message)

client.run("automate", on_event=on_event)

When no callback is given, events fall back to print + the driverclient logger, so the terminal/log experience is unchanged.

Each ClientEvent (frozen dataclass, ev.to_dict() for transport) has:

field	type	meaning
phase	str	scan resolve install capture windows_update dump upload pipeline done
status	str	start progress ok warn error done
message	str	human-readable line (what the fallback prints)
current	int | None	counter position (e.g. driver 3 of 12)
total	int | None	counter total
percent	float | None	0–100 for long single operations
data	dict	structured specifics (driver_name, hwid, path, error, …)
ts	float	time.time() at emit

phase and status are a fixed, documented vocabulary — this is the interface a GUI binds to; treat changes to it as an API change.

Events are emitted on whatever thread the op is running on — and the parallel download/export/upload pools mean some events arrive on worker threads. The Qt pattern below (queued cross-thread signals) is safe regardless.

PyQt consumer pattern

Run the (minutes-long, blocking) client.run(...) on a QThread, never the UI thread. The on_event callback must not touch widgets — it emits a Qt signal carrying event.to_dict(); a slot on the UI thread updates the widgets. Qt delivers cross-thread signals via the receiver's event loop (queued), which is the thread-safe way to drive the UI.

from PyQt6.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from driverclient import DriverClient


class Worker(QObject):
    event    = pyqtSignal(dict)     # carries ClientEvent.to_dict()
    finished = pyqtSignal(object)   # carries the op's result

    def __init__(self, command):
        super().__init__()
        self.command = command

    @pyqtSlot()
    def run(self):
        client = DriverClient(local_repo_url="http://REPO_HOST:8000",
                              node_key="YOUR_NODE_KEY")
        # Called on THIS worker thread — only emit a signal, never touch widgets.
        result = client.run(self.command, on_event=lambda ev: self.event.emit(ev.to_dict()))
        self.finished.emit(result)


class Panel:
    """Wire a worker onto a thread and bind its signals to UI-thread slots."""
    def start(self, command):
        self.thread = QThread()
        self.worker = Worker(command)
        self.worker.moveToThread(self.thread)
        self.thread.started.connect(self.worker.run)
        self.worker.event.connect(self.on_event)        # queued → runs on UI thread
        self.worker.finished.connect(self.on_finished)
        self.thread.start()

    @pyqtSlot(dict)
    def on_event(self, ev):
        # Safe: this runs on the UI thread. Update widgets here.
        self.status_label.setText(ev["message"])
        if ev["phase"] == "install" and ev["status"] in ("ok", "error"):
            self.step_list.addItem(ev["message"])
        if ev["total"] and ev["current"] is not None:
            self.progress_bar.setMaximum(ev["total"])
            self.progress_bar.setValue(ev["current"])
        elif ev["percent"] is not None:
            self.progress_bar.setValue(int(ev["percent"]))

    @pyqtSlot(object)
    def on_finished(self, result):
        self.thread.quit()
        self.thread.wait()

Requirements

• Python >= 3.10
• Windows target for the actual driver operations (pnputil, WMI, DISM).