From 2467a24a8eb45517fe1c525fc12ce89c40142144 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 22 Jul 2026 18:21:39 -0400 Subject: [PATCH] Add PowerStrip device family, PwrUSBDevice driver, and HIDPort transport Problem: A PwrUSB / PowerUSB power bar (USB HID 04d8:003f, enumerates as "Simple HID Device Demo") with three switchable outlets was attached, but PyHardwareLibrary had no power-strip family or driver for it, and no way to reach an HID device the OS claims (on macOS, IOHIDFamily owns the interface, so neither SerialPort nor USBPort/libusb can open it). Solution: Add hardwarelibrary/powerstrips/ following the interface-segregated capability-mixin pattern of sources/ and daq/. PowerStripDevice is a thin marker base over PhysicalDevice exposing capabilities()/hasCapability(); capabilities.py defines OutletSwitchingControl, DefaultOutletControl, and CurrentMeteringControl, each a public method over a do* hook. PwrUSBDevice implements all three, speaking the device's single-byte HID report protocol through a HIDPort. Outlet state is cached on write because live readback is unreliable on this firmware. DebugPwrUSBPort decodes the real command bytes so DebugPwrUSBDevice exercises the encoding end to end without hardware. Add HIDPort, a CommunicationPort backed by hidapi (IOKit on macOS), alongside SerialPort and USBPort. It is required because the OS claims the HID interface; hidapi is optional, installed via the new pwrusb extra (pip install -e .[pwrusb]), whose package is hidapi (cython-hidapi, imports as hid), not the different hid package. HIDPort.flush drains with a strictly positive per-read timeout, since hidapi blocks on timeout_ms=0. The single-byte protocol was reverse-engineered publicly and cross-checked against aarossig/pwrusbctl (Apache-2.0) and pwrusb.com; the implementation is our own (attribution in the module docstring and CHANGELOG). Verified end to end on the lab unit (a PowerUSB "Basic"): opens over HID and switches outlets 1/2/3 on and off. Metering is only meaningful on a "Smart" unit. Documents the new family in the pyhardwarelibrary skill. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/skills/pyhardwarelibrary/SKILL.md | 45 +++- CHANGELOG.md | 21 ++ hardwarelibrary/communication/__init__.py | 1 + hardwarelibrary/communication/hidport.py | 98 +++++++++ hardwarelibrary/powerstrips/__init__.py | 6 + hardwarelibrary/powerstrips/capabilities.py | 99 +++++++++ .../powerstrips/powerstripdevice.py | 23 ++ hardwarelibrary/powerstrips/pwrusb.py | 207 ++++++++++++++++++ hardwarelibrary/tests/testPwrUSB.py | 124 +++++++++++ pyproject.toml | 6 + 10 files changed, 629 insertions(+), 1 deletion(-) create mode 100644 hardwarelibrary/communication/hidport.py create mode 100644 hardwarelibrary/powerstrips/__init__.py create mode 100644 hardwarelibrary/powerstrips/capabilities.py create mode 100644 hardwarelibrary/powerstrips/powerstripdevice.py create mode 100644 hardwarelibrary/powerstrips/pwrusb.py create mode 100644 hardwarelibrary/tests/testPwrUSB.py diff --git a/.claude/skills/pyhardwarelibrary/SKILL.md b/.claude/skills/pyhardwarelibrary/SKILL.md index 1ba0ff2..93a1154 100644 --- a/.claude/skills/pyhardwarelibrary/SKILL.md +++ b/.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. diff --git a/CHANGELOG.md b/CHANGELOG.md index af8892e..267a532 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index d9d11b9..4512c98 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/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 diff --git a/hardwarelibrary/communication/hidport.py b/hardwarelibrary/communication/hidport.py new file mode 100644 index 0000000..22ee8c3 --- /dev/null +++ b/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) diff --git a/hardwarelibrary/powerstrips/__init__.py b/hardwarelibrary/powerstrips/__init__.py new file mode 100644 index 0000000..95c6121 --- /dev/null +++ b/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 diff --git a/hardwarelibrary/powerstrips/capabilities.py b/hardwarelibrary/powerstrips/capabilities.py new file mode 100644 index 0000000..9df781f --- /dev/null +++ b/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): + ... diff --git a/hardwarelibrary/powerstrips/powerstripdevice.py b/hardwarelibrary/powerstrips/powerstripdevice.py new file mode 100644 index 0000000..1d754a8 --- /dev/null +++ b/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) diff --git a/hardwarelibrary/powerstrips/pwrusb.py b/hardwarelibrary/powerstrips/pwrusb.py new file mode 100644 index 0000000..95bc83c --- /dev/null +++ b/hardwarelibrary/powerstrips/pwrusb.py @@ -0,0 +1,207 @@ +"""PwrUSB / PowerUSB controllable power strip (USB HID, 04d8:003f). + +The strip enumerates as a Microchip HID device ("Simple HID Device Demo") and +carries one always-on outlet plus three switchable outlets. It is driven with a +single-byte HID report protocol: write one command byte; for query commands, +read the reply back. + +Command bytes (outlets are 1-based here to match the physical labels): + + Action Outlet 1 Outlet 2 Outlet 3 + On 0x41 0x43 0x45 + Off 0x42 0x44 0x50 + Default on 0x4E 0x47 0x4F (state after mains power returns) + Default off 0x46 0x51 0x48 + + Query Byte Reply + Device type 0xAA 1 byte (1=Basic 2=Digital IO 3=Watchdog 4=Smart) + Current 0xB1 2 bytes big-endian, milliamps + Charge 0xB2 4 bytes big-endian, milliamp-minutes + Reset charge 0xB3 none + +Live outlet-state readback is unreliable on this hardware (a documented firmware +quirk: the read tends to return only the last port read), so the driver caches +each outlet's state as it is written and reports the cache. + +The driver talks to the device through a HIDPort (hidapi/IOKit). HID is required +because the OS claims the interface: on macOS neither SerialPort (no /dev node) +nor USBPort/libusb can open it. Install the optional 'pwrusb' extra for hidapi. +The single-byte protocol above was reverse-engineered publicly; it was cross- +checked against aarossig/pwrusbctl (Apache-2.0, https://github.com/aarossig/pwrusbctl) +and pwrusb.com. The implementation here is our own. +""" + +from hardwarelibrary.communication.hidport import HIDPort +from hardwarelibrary.communication.debugport import DebugPort +from hardwarelibrary.powerstrips.powerstripdevice import PowerStripDevice +from hardwarelibrary.powerstrips.capabilities import ( + OutletSwitchingControl, DefaultOutletControl, CurrentMeteringControl, +) + + +class PwrUSBDevice(PowerStripDevice, OutletSwitchingControl, + DefaultOutletControl, CurrentMeteringControl): + classIdVendor = 0x04d8 + classIdProduct = 0x003f + + outletCommands = { + "on": [0x41, 0x43, 0x45], + "off": [0x42, 0x44, 0x50], + "defaultOn": [0x4E, 0x47, 0x4F], + "defaultOff": [0x46, 0x51, 0x48], + } + getDeviceTypeCommand = 0xAA + getCurrentCommand = 0xB1 + getChargeCommand = 0xB2 + resetChargeCommand = 0xB3 + + deviceTypes = {1: "Basic", 2: "Digital IO", 3: "Watchdog", 4: "Smart"} + + switchableOutletCount = 3 + + def __init__(self, serialNumber: str = None, idProduct: int = None, + idVendor: int = None, portPath: str = None): + self.portPath = portPath + self._outletStateCache = [False] * self.switchableOutletCount + super().__init__(serialNumber=serialNumber, idProduct=idProduct, idVendor=idVendor) + self.port = None + + def doInitializeDevice(self): + # Let the underlying exception propagate unwrapped: PhysicalDevice. + # initializeDevice wraps it once in UnableToInitialize, so callers can + # read the true cause (e.g. an OSError from hidapi) from its args. + try: + if self.portPath == "debug": + self.port = DebugPwrUSBPort() + else: + self.port = HIDPort(idVendor=self.idVendor, idProduct=self.idProduct, + serialNumber=self.serialNumber) + self.port.open() + + self._outletStateCache = [False] * self.switchableOutletCount + except Exception: + if self.port is not None and self.port.isOpen: + self.port.close() + self.port = None + raise + + def doShutdownDevice(self): + if self.port is not None: + self.port.close() + self.port = None + + def _validateOutlet(self, outlet: int): + if outlet not in range(1, self.switchableOutletCount + 1): + raise ValueError("Outlet must be 1..{0}, got {1}".format( + self.switchableOutletCount, outlet)) + + def deviceType(self) -> str: + # flush() first so the reply is read from a clean buffer: a report can be + # longer than the bytes we need, leaving a remainder buffered from the + # previous query. + self.port.flush() + self.port.writeData(bytearray([self.getDeviceTypeCommand])) + reply = self.port.readData(length=1) + return self.deviceTypes.get(reply[0], "Unknown") + + def doGetOutletCount(self) -> int: + return self.switchableOutletCount + + def doSetOutletState(self, outlet: int, isOn: bool): + self._validateOutlet(outlet) + key = "on" if isOn else "off" + self.port.writeData(bytearray([self.outletCommands[key][outlet - 1]])) + self._outletStateCache[outlet - 1] = bool(isOn) + + def doGetOutletState(self, outlet: int) -> bool: + self._validateOutlet(outlet) + return self._outletStateCache[outlet - 1] + + def doSetOutletDefaultState(self, outlet: int, isOn: bool): + self._validateOutlet(outlet) + key = "defaultOn" if isOn else "defaultOff" + self.port.writeData(bytearray([self.outletCommands[key][outlet - 1]])) + + def doGetCurrent(self) -> float: + self.port.flush() + self.port.writeData(bytearray([self.getCurrentCommand])) + reply = self.port.readData(length=2) + milliamps = (reply[0] << 8) | reply[1] + return milliamps / 1000.0 + + def doGetAccumulatedCharge(self) -> float: + self.port.flush() + self.port.writeData(bytearray([self.getChargeCommand])) + reply = self.port.readData(length=4) + milliampMinutes = (reply[0] << 24) | (reply[1] << 16) | (reply[2] << 8) | reply[3] + return milliampMinutes / 60000.0 + + def doResetAccumulatedCharge(self): + self.port.writeData(bytearray([self.resetChargeCommand])) + + +class DebugPwrUSBPort(DebugPort): + """Mock port that decodes PwrUSBDevice command bytes and answers queries. + + It interprets the same command bytes the real strip does (so the driver's + encoding is exercised end to end) and keeps in-memory outlet/default/meter + state that tests can inspect. + """ + + def __init__(self): + super().__init__() + self.outletStates = [False] * PwrUSBDevice.switchableOutletCount + self.defaultStates = [False] * PwrUSBDevice.switchableOutletCount + self.currentMilliamps = 250 + self.chargeMilliampMinutes = 0 + self.deviceTypeCode = 4 # Smart: the metering-capable model + + def processInputBuffers(self, endPointIndex): + inputBytes = self.inputBuffers[endPointIndex] + if len(inputBytes) == 0: + return + + byte = inputBytes[0] + self.inputBuffers[endPointIndex] = bytearray() + + commands = PwrUSBDevice.outletCommands + for index in range(PwrUSBDevice.switchableOutletCount): + if byte == commands["on"][index]: + self.outletStates[index] = True + return + if byte == commands["off"][index]: + self.outletStates[index] = False + return + if byte == commands["defaultOn"][index]: + self.defaultStates[index] = True + return + if byte == commands["defaultOff"][index]: + self.defaultStates[index] = False + return + + if byte == PwrUSBDevice.getDeviceTypeCommand: + self.writeToOutputBuffer(bytearray([self.deviceTypeCode]), endPointIndex) + elif byte == PwrUSBDevice.getCurrentCommand: + value = self.currentMilliamps + self.writeToOutputBuffer(bytearray([(value >> 8) & 0xFF, value & 0xFF]), endPointIndex) + elif byte == PwrUSBDevice.getChargeCommand: + value = self.chargeMilliampMinutes + self.writeToOutputBuffer(bytearray([ + (value >> 24) & 0xFF, (value >> 16) & 0xFF, + (value >> 8) & 0xFF, value & 0xFF]), endPointIndex) + elif byte == PwrUSBDevice.resetChargeCommand: + self.chargeMilliampMinutes = 0 + else: + print("Unrecognized PwrUSB command byte: 0x{0:02x}".format(byte)) + + +class DebugPwrUSBDevice(PwrUSBDevice): + """Hardware-free PwrUSB strip for tests, backed by DebugPwrUSBPort.""" + + classIdVendor = 0xFFFF + classIdProduct = 0xFFF9 + + def __init__(self, serialNumber: str = "debug"): + super().__init__(serialNumber=serialNumber, + idProduct=self.classIdProduct, idVendor=self.classIdVendor, + portPath="debug") diff --git a/hardwarelibrary/tests/testPwrUSB.py b/hardwarelibrary/tests/testPwrUSB.py new file mode 100644 index 0000000..1fe9502 --- /dev/null +++ b/hardwarelibrary/tests/testPwrUSB.py @@ -0,0 +1,124 @@ +import env +import unittest + +from hardwarelibrary.physicaldevice import PhysicalDevice +from hardwarelibrary.powerstrips import ( + PwrUSBDevice, DebugPwrUSBDevice, + OutletSwitchingControl, DefaultOutletControl, CurrentMeteringControl, +) + + +class TestDebugPwrUSBDevice(unittest.TestCase): + def setUp(self): + self.device = DebugPwrUSBDevice() + self.device.initializeDevice() + + def tearDown(self): + self.device.shutdownDevice() + + def testCreate(self): + self.assertIsNotNone(self.device) + + def testInitializeAndShutdown(self): + device = DebugPwrUSBDevice() + device.initializeDevice() + device.shutdownDevice() + + def testOutletCount(self): + self.assertEqual(self.device.outletCount, 3) + + def testCapabilities(self): + capabilities = self.device.capabilities() + self.assertIn(OutletSwitchingControl, capabilities) + self.assertIn(DefaultOutletControl, capabilities) + self.assertIn(CurrentMeteringControl, capabilities) + + def testHasCapability(self): + self.assertTrue(self.device.hasCapability(OutletSwitchingControl)) + self.assertTrue(self.device.hasCapability(CurrentMeteringControl)) + + def testAllOutletsStartOff(self): + for outlet in (1, 2, 3): + self.assertFalse(self.device.isOutletOn(outlet)) + + def testTurnOutletOnAndOff(self): + for outlet in (1, 2, 3): + self.device.turnOutletOn(outlet) + self.assertTrue(self.device.isOutletOn(outlet)) + self.device.turnOutletOff(outlet) + self.assertFalse(self.device.isOutletOn(outlet)) + + def testSetOutletState(self): + self.device.setOutletState(2, True) + self.assertTrue(self.device.isOutletOn(2)) + self.assertFalse(self.device.isOutletOn(1)) + self.assertFalse(self.device.isOutletOn(3)) + + def testSwitchingSendsCorrectBytesPerOutlet(self): + # The mock port decodes the real command bytes; confirm each outlet's + # ON/OFF maps to exactly that outlet on the wire. + self.device.turnOutletOn(3) + self.assertEqual(self.device.port.outletStates, [False, False, True]) + + def testDefaultStateSetters(self): + self.device.setOutletDefaultOn(1) + self.device.setOutletDefaultOff(2) + self.assertEqual(self.device.port.defaultStates, [True, False, False]) + + def testInvalidOutletRaises(self): + for outlet in (0, 4, -1): + with self.assertRaises(ValueError): + self.device.turnOutletOn(outlet) + with self.assertRaises(ValueError): + self.device.isOutletOn(outlet) + + def testDeviceType(self): + self.assertEqual(self.device.deviceType(), "Smart") + + def testGetCurrent(self): + current = self.device.current() + self.assertAlmostEqual(current, 0.250) + + def testAccumulatedChargeAndReset(self): + self.device.port.chargeMilliampMinutes = 120000 + self.assertAlmostEqual(self.device.accumulatedCharge(), 2.0) + self.device.resetAccumulatedCharge() + self.assertAlmostEqual(self.device.accumulatedCharge(), 0.0) + + +class TestPwrUSBDevice(unittest.TestCase): + def setUp(self): + super().setUp() + self.device = PwrUSBDevice() + try: + self.device.initializeDevice() + except PhysicalDevice.UnableToInitialize as error: + cause = error.args[0] if error.args else None + # No strip attached raises an OSError from hidapi; a host without the + # optional hidapi dependency (e.g. CI, which installs only [dev]) + # raises ImportError. Either way there is no hardware to exercise, so + # skip rather than fail. + if isinstance(cause, (OSError, ImportError)): + self.skipTest("No PwrUSB connected (or hidapi not installed)") + raise + + def tearDown(self): + super().tearDown() + self.device.shutdownDevice() + + def testOutletCount(self): + self.assertEqual(self.device.outletCount, 3) + + def testTurnOutletOnAndOff(self): + self.device.turnOutletOn(1) + self.assertTrue(self.device.isOutletOn(1)) + self.device.turnOutletOff(1) + self.assertFalse(self.device.isOutletOn(1)) + + def testDeviceType(self): + self.assertIn(self.device.deviceType(), + ("Basic", "Digital IO", "Watchdog", "Smart", "Unknown")) + + +if __name__ == '__main__': + unittest.main() diff --git a/pyproject.toml b/pyproject.toml index 963182f..8ff1b8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,12 @@ thorlabs = [ # Backend for ThorlabsKinesisDevice (the Thorlabs Kinesis runtime wrapper). "pylablib", ] +pwrusb = [ + # HID transport (HIDPort) for PwrUSBDevice. Imports as `hid`. Required to + # reach the strip on macOS, where the OS claims the HID interface and neither + # SerialPort nor USBPort/libusb can open it. + "hidapi", +] [project.urls] Homepage = "https://github.com/DCC-Lab/PyHardwareLibrary"