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"