Merge pull request #115 from DCC-Lab/pwrusb-power-strip

Add PowerStrip device family, PwrUSBDevice driver, and HIDPort transport

Author: dccote

.claude/skills/pyhardwarelibrary/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: pyhardwarelibrary
-description: Control lab hardware with the PyHardwareLibrary Python package — motion stages, spectrometers, lasers, power meters, DAQs, and oscilloscopes. Use when a user wants to move a stage, acquire a spectrum, turn a laser on/off or set its power/wavelength, read a power meter, read/write DAQ voltages, discover connected devices, run code without hardware (debug devices), or build a headless/GUI controller around a device. Covers the device lifecycle, per-family public API, event notifications, and the DeviceController worker-thread wrapper.
+description: Control lab hardware with the PyHardwareLibrary Python package — motion stages, spectrometers, lasers, power meters, DAQs, oscilloscopes, and USB power strips. Use when a user wants to move a stage, acquire a spectrum, turn a laser on/off or set its power/wavelength, read a power meter, read/write DAQ voltages, switch power-strip outlets on/off, discover connected devices, run code without hardware (debug devices), or build a headless/GUI controller around a device. Covers the device lifecycle, per-family public API, event notifications, and the DeviceController worker-thread wrapper.
 ---
 
 # Using PyHardwareLibrary
@@ -10,6 +10,15 @@ subclass with a uniform lifecycle and a small, predictable public API per family
 skill is for *using* devices to get work done (move, measure, set). To *write a new
 driver*, read `CLAUDE.md` and `README-4-New-device-coding-example.md` instead.
 
+**Use the primitives from `CommunicationPort` for all communications.** A driver's `do*`
+methods talk to hardware only through `self.port` — `writeData` / `readData` (and the
+`readString` / `writeString` / `writeStringReadMatch` helpers built on them). Do **not**
+invent new send/receive helpers on the device (no `_sendCommandByte`, no `_query`); call
+the port methods directly. To support a new transport (serial, USB, HID, TCP, ...),
+subclass `CommunicationPort` and go through the expected mechanism of **overriding
+`readData` and `writeData`** (plus `open` / `close` / `isOpen` / `flush`) — the string and
+transaction helpers then compose on top for free. `HIDPort` is the reference example.
+
 ## The one rule: lifecycle
 
 Every device follows the same pattern. Construct it, initialize it, use it, shut it down.
@@ -185,6 +194,40 @@ daq.setDigitalValue(1, channel=5)
 daq.shutdownDevice()
 ```
 
+### Power strips — PwrUSB / PowerUSB
+
+Base: `PowerStripDevice` plus capability mixins (like lasers/DAQ). Outlets are addressed
+**1-based** to match the physical labels. A device only has the methods of the capabilities
+it declares.
+
+| Capability | Methods |
+|---|---|
+| `OutletSwitchingControl` | `turnOutletOn(n)`, `turnOutletOff(n)`, `setOutletState(n, isOn)`, `isOutletOn(n)`, `outletCount` |
+| `DefaultOutletControl` | `setOutletDefaultOn(n)`, `setOutletDefaultOff(n)`, `setOutletDefaultState(n, isOn)` |
+| `CurrentMeteringControl` | `current()` (A), `accumulatedCharge()` (Ah), `resetAccumulatedCharge()` |
+
+```python
+from hardwarelibrary.powerstrips import PwrUSBDevice
+
+strip = PwrUSBDevice()             # first strip found (over hidapi)
+strip.initializeDevice()
+strip.turnOutletOn(1)             # outlets are 1, 2, 3
+print(strip.isOutletOn(1), "of", strip.outletCount)
+strip.setOutletDefaultOff(1)      # boot state after mains power returns
+strip.turnOutletOff(1)
+strip.shutdownDevice()
+```
+
+- **USB HID only** (`04d8:003f`), driven over hidapi (`HIDPort`). The OS claims the HID
+  interface (on macOS `SerialPort` has no `/dev` node and `USBPort`/libusb can't open it),
+  so control needs the hidapi extra: `pip install -e .[pwrusb]` (PyPI package `hidapi`,
+  imports as `hid` — *not* the different `hid` package).
+- **Outlet state is cached on write** (live readback is unreliable on this firmware), so
+  `isOutletOn(n)` returns the last commanded state.
+- **Metering is only meaningful on the "Smart" model**; a "Basic" unit returns
+  non-measurement values from `current()` / `accumulatedCharge()`.
+- No hardware? `DebugPwrUSBDevice()` emulates the protocol in memory.
+
 ### Oscilloscopes — Tektronix TDS
 
 Instantiated directly (no family/driver split); methods are SCPI per instrument.

CHANGELOG.md

@@ -7,6 +7,27 @@ API changes can land even when the minor version is unchanged.
 ## [Unreleased]
 
 ### Added
+- PowerStrip device family (`hardwarelibrary/powerstrips/`) plus its first driver
+  `PwrUSBDevice` (and `DebugPwrUSBDevice`) for the PwrUSB / PowerUSB controllable
+  power strip (USB HID `04d8:003f`, enumerates as "Simple HID Device Demo"). The
+  family follows the interface-segregated capability-mixin pattern used by
+  `sources/` and `daq/`: `PowerStripDevice` is a thin marker base over
+  `PhysicalDevice`, and behaviour comes from `OutletSwitchingControl`
+  (`turnOutletOn`/`turnOutletOff`/`setOutletState`/`isOutletOn`/`outletCount`,
+  outlets 1-based), `DefaultOutletControl` (per-outlet power-on default state),
+  and `CurrentMeteringControl` (`current()` in A, `accumulatedCharge()` in Ah,
+  `resetAccumulatedCharge()`). The strip speaks a single-byte HID report protocol
+  driven through a `HIDPort`; the protocol was reverse-engineered publicly and
+  cross-checked against aarossig/pwrusbctl (Apache-2.0) and pwrusb.com, but the
+  implementation is our own. Outlet state is cached on write because live
+  readback is unreliable on this firmware.
+- `HIDPort` (`hardwarelibrary/communication/hidport.py`): a `CommunicationPort`
+  over a USB HID device, backed by hidapi (IOKit on macOS), alongside
+  `SerialPort` and `USBPort`. Needed because an HID device the OS claims has no
+  `/dev` node (so `SerialPort` cannot reach it) and cannot be claimed by libusb
+  (so `USBPort` cannot either) -- notably on macOS, where `IOHIDFamily` owns the
+  interface. hidapi is an optional dependency; install it with the new `pwrusb`
+  extra (`pip install -e .[pwrusb]`), required to drive the strip.
 - `VerdiGDevice` (and `DebugVerdiGDevice`): a laser-source driver for the Coherent
   "HOPS" (High Output Power Supply) laser -- Genesis heads / Verdi G-C, e.g. the
   lab Genesis CX-Vis (head `G532`). A HOPS supply is not a serial device: its

hardwarelibrary/communication/__init__.py

@@ -4,6 +4,7 @@
 from .tcpport import TCPPort, UnableToOpenTCPPort
 from .labviewtcpport import LabviewTCPPort
 from .usbport import USBPort
+from .hidport import HIDPort
 from .diagnostics import USBParameters, DeviceCommand, USBDeviceDescription
 from .debugport import DebugPort, TableDrivenDebugPort
 import usb.backend.libusb1

hardwarelibrary/communication/hidport.py

@@ -0,0 +1,98 @@
+from .communicationport import CommunicationPort, CommunicationReadTimeout
+
+
+class HIDPort(CommunicationPort):
+    """A CommunicationPort over a USB HID device, backed by hidapi.
+
+    HID is the only way to reach a device the operating system claims as HID.
+    On macOS the IOHIDFamily driver owns the interface, so neither SerialPort
+    (there is no /dev node) nor USBPort (libusb cannot claim the interface) can
+    open it; hidapi goes through the same IOKit HID manager the OS reserves.
+
+    This exposes the same primitives as SerialPort/USBPort (open/close/
+    readData/writeData/flush), so a driver written against CommunicationPort
+    works unchanged. It requires the optional 'hidapi' dependency (imported as
+    'hid'); the import is deferred to open() so the rest of the library does not
+    depend on it.
+
+    hidapi frames every report with the report ID as byte 0. Devices that use a
+    single unnumbered report expect that byte to be 0x00; writeData prepends it,
+    and the count it returns excludes it.
+    """
+
+    def __init__(self, idVendor, idProduct, serialNumber=None):
+        super().__init__()
+        self.idVendor = idVendor
+        self.idProduct = idProduct
+        self.serialNumber = serialNumber
+        self.device = None
+        self.defaultTimeout = 500
+        self.reportSize = 64
+        # Bounded drain for flush(): a small positive per-read timeout (never 0,
+        # which blocks in hidapi rather than returning immediately) and a cap on
+        # the number of reports drained, so flush can never hang.
+        self.drainTimeout = 2
+        self.maxDrainReports = 16
+        self._internalBuffer = bytearray()
+
+    @property
+    def isOpen(self) -> bool:
+        return self.device is not None
+
+    def open(self):
+        if self.isOpen:
+            raise Exception("Port already open")
+
+        import hid
+        self.device = hid.device()
+        if self.serialNumber in (None, "*", ".*"):
+            self.device.open(self.idVendor, self.idProduct)
+        else:
+            self.device.open(self.idVendor, self.idProduct, self.serialNumber)
+        self._internalBuffer = bytearray()
+
+    def close(self):
+        with self.portLock:
+            if self.device is not None:
+                self.device.close()
+                self.device = None
+            self._internalBuffer = bytearray()
+
+    def bytesAvailable(self, endPoint=None) -> int:
+        with self.portLock:
+            return len(self._internalBuffer)
+
+    def flush(self, endPoint=None):
+        with self.portLock:
+            self._internalBuffer = bytearray()
+            if self.device is None:
+                return
+            # timeout_ms must stay positive: hidapi blocks on 0. The cap bounds
+            # the wait even if a device were to stream input reports.
+            for _ in range(self.maxDrainReports):
+                if not self.device.read(self.reportSize, timeout_ms=self.drainTimeout):
+                    break
+
+    def readData(self, length, endPoint=None) -> bytearray:
+        with self.portLock:
+            while length > len(self._internalBuffer):
+                chunk = self.device.read(self.reportSize, timeout_ms=self.defaultTimeout)
+                if not chunk:
+                    raise CommunicationReadTimeout(
+                        "Read {0} of {1} requested bytes from HID device".format(
+                            len(self._internalBuffer), length))
+                self._internalBuffer += bytearray(chunk)
+
+            data = self._internalBuffer[:length]
+            self._internalBuffer = self._internalBuffer[length:]
+
+        return data
+
+    def writeData(self, data, endPoint=None) -> int:
+        with self.portLock:
+            # Byte 0 is the HID report ID (0x00 for a single, unnumbered report).
+            written = self.device.write(bytes([0x00]) + bytes(data))
+            if written < 0:
+                raise IOError("Unable to write to HID device")
+
+        return max(0, written - 1)

hardwarelibrary/powerstrips/__init__.py

@@ -0,0 +1,6 @@
+from .capabilities import (
+    Capability,
+    OutletSwitchingControl, DefaultOutletControl, CurrentMeteringControl,
+)
+from .powerstripdevice import PowerStripDevice
+from .pwrusb import PwrUSBDevice, DebugPwrUSBDevice, DebugPwrUSBPort

hardwarelibrary/powerstrips/capabilities.py

@@ -0,0 +1,99 @@
+from abc import ABC, abstractmethod
+
+
+class Capability(ABC):
+    pass
+
+
+class OutletSwitchingControl(Capability):
+    """Switch individual outlets on and off and read their state.
+
+    Outlets are addressed by their physical label (1-based): the first
+    switchable outlet is outlet 1. Some strips also carry an always-on outlet
+    that is not switchable and is not counted here.
+    """
+
+    def turnOutletOn(self, outlet: int):
+        self.doSetOutletState(outlet, True)
+
+    def turnOutletOff(self, outlet: int):
+        self.doSetOutletState(outlet, False)
+
+    def setOutletState(self, outlet: int, isOn: bool):
+        self.doSetOutletState(outlet, isOn)
+
+    def isOutletOn(self, outlet: int) -> bool:
+        return self.doGetOutletState(outlet)
+
+    @property
+    def outletCount(self) -> int:
+        return self.doGetOutletCount()
+
+    @abstractmethod
+    def doSetOutletState(self, outlet: int, isOn: bool):
+        ...
+
+    @abstractmethod
+    def doGetOutletState(self, outlet: int) -> bool:
+        ...
+
+    @abstractmethod
+    def doGetOutletCount(self) -> int:
+        ...
+
+
+class DefaultOutletControl(Capability):
+    """Set the power-on (boot) state of individual outlets.
+
+    Distinct from OutletSwitchingControl: this configures the state each outlet
+    powers up in after the strip loses and regains mains power, not its state
+    right now.
+    """
+
+    def setOutletDefaultOn(self, outlet: int):
+        self.doSetOutletDefaultState(outlet, True)
+
+    def setOutletDefaultOff(self, outlet: int):
+        self.doSetOutletDefaultState(outlet, False)
+
+    def setOutletDefaultState(self, outlet: int, isOn: bool):
+        self.doSetOutletDefaultState(outlet, isOn)
+
+    @abstractmethod
+    def doSetOutletDefaultState(self, outlet: int, isOn: bool):
+        ...
+
+
+class CurrentMeteringControl(Capability):
+    """Measure the strip's total current draw and accumulated charge.
+
+    Only metering-capable strips (e.g. the PowerUSB "Smart" model) implement
+    this; a driver mixes it in only when the hardware supports it. Values are in
+    SI units at this boundary: current in amperes, accumulated charge in
+    ampere-hours.
+    """
+
+    unit = "A"
+    isReadable = True
+    isWritable = False
+
+    def current(self) -> float:
+        return self.doGetCurrent()
+
+    def accumulatedCharge(self) -> float:
+        return self.doGetAccumulatedCharge()
+
+    def resetAccumulatedCharge(self):
+        self.doResetAccumulatedCharge()
+
+    @abstractmethod
+    def doGetCurrent(self) -> float:
+        ...
+
+    @abstractmethod
+    def doGetAccumulatedCharge(self) -> float:
+        ...
+
+    @abstractmethod
+    def doResetAccumulatedCharge(self):
+        ...

hardwarelibrary/powerstrips/powerstripdevice.py

@@ -0,0 +1,23 @@
+from hardwarelibrary.physicaldevice import PhysicalDevice
+from hardwarelibrary.powerstrips.capabilities import Capability
+
+
+class PowerStripDevice(PhysicalDevice):
+    """Family marker base for controllable power strips (switched outlets).
+
+    A driver subclasses this and mixes in the capability classes from
+    capabilities.py (OutletSwitchingControl, DefaultOutletControl,
+    CurrentMeteringControl). The behaviour lives in the mixins; this base only
+    reports which capabilities a given driver actually implements.
+    """
+
+    def capabilities(self) -> list:
+        # The capability mixins, not the marker nor the device class itself
+        # (a driver is a Capability subclass too, but it is a PhysicalDevice).
+        return [klass for klass in type(self).__mro__
+                if issubclass(klass, Capability)
+                and klass is not Capability
+                and not issubclass(klass, PhysicalDevice)]
+
+    def hasCapability(self, capabilityClass) -> bool:
+        return isinstance(self, capabilityClass)