arrdem/source
1
0
Fork 0
Code Issues 11 Pull requests Projects Releases Packages Wiki Activity

Fmt.

Browse source
This commit is contained in:
Reid 'arrdem' McKenzie 2021-11-09 12:31:31 -07:00
parent 8707dd3142
commit d49034c0fa
4 changed files with 90 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
projects/clusterctrl/src/python/clusterctrl/__main__.py
View file

@ -31,6 +31,7 @@ import click
# #
# <order> selects which Cluster CTRL devices matches that <order> number # <order> selects which Cluster CTRL devices matches that <order> number
@click.group() @click.group()
def cli(): def cli():
pass pass

1
projects/clusterctrl/src/python/clusterctrl/cmd/power.py
View file

@ -9,6 +9,7 @@ import click
def power(): def power():
pass pass
@power.command("on") @power.command("on")
@click.argument("devices", nargs="*") @click.argument("devices", nargs="*")
def power_on(devices): def power_on(devices):

52
projects/clusterctrl/src/python/clusterctrl/driver.py
View file

@ -42,6 +42,7 @@ def once(f):
""" """
unset = val = object() unset = val = object()
def _helper(*args, **kwargs): def _helper(*args, **kwargs):
nonlocal val nonlocal val
if val is unset: if val is unset:
@ -56,13 +57,13 @@ I2C_ADDRESS = 0x20
class BoardType(Enum): class BoardType(Enum):
DA = 0x00 # Unknown, presunably a prototype DA = 0x00 # Unknown, presunably a prototype
SINGLE = 0x01 # Do the 'single' and 'triple' really use the same board ID? SINGLE = 0x01 # Do the 'single' and 'triple' really use the same board ID?
TRIPLE = 0x01 # ??? TRIPLE = 0x01 # ???
PHAT = 0x02 # The ClusterHAT boards PHAT = 0x02 # The ClusterHAT boards
CTRL = 0x02 # The ClusterCTRL boards CTRL = 0x02 # The ClusterCTRL boards
A6 = 0x03 # https://clusterctrl.com/p/aplus6 A6 = 0x03 # https://clusterctrl.com/p/aplus6
STACK = 0x04 # https://shop.pimoroni.com/products/clusterctrl-stack STACK = 0x04 # https://shop.pimoroni.com/products/clusterctrl-stack
class Status(Enum): class Status(Enum):
@ -105,8 +106,8 @@ class Cmd(Enum):
ALERT_ON = 0x05 # Turn on Alert LED ALERT_ON = 0x05 # Turn on Alert LED
ALERT_OFF = 0x06 # Turn off Alert LED ALERT_OFF = 0x06 # Turn off Alert LED
HUB_CYCLE = 0x07 # Reset USB HUB (turn off for data0*10ms, then back on) HUB_CYCLE = 0x07 # Reset USB HUB (turn off for data0*10ms, then back on)
HUB_ON = 0x08 # Turn on the USB hub HUB_ON = 0x08 # Turn on the USB hub
HUB_OFF = 0x09 # Turn off the USB hub HUB_OFF = 0x09 # Turn off the USB hub
LED_EN = 0x0A # Enable Px LED (data0=x) (PHAT only) LED_EN = 0x0A # Enable Px LED (data0=x) (PHAT only)
LED_DIS = 0x0B # Disable Px LED (data0=x) (PHAT only) LED_DIS = 0x0B # Disable Px LED (data0=x) (PHAT only)
PWR_ON = 0x0C # Turn off PWR LED PWR_ON = 0x0C # Turn off PWR LED
@ -146,6 +147,7 @@ class PiRef(NamedTuple):
These IDs are expected to be unique at the host level; not at the cluster level. These IDs are expected to be unique at the host level; not at the cluster level.
""" """
controller_id: int controller_id: int
pi_id: int pi_id: int
@ -154,11 +156,13 @@ class PiRef(NamedTuple):
class ClusterCTRLv2Driver(object): class ClusterCTRLv2Driver(object):
def __init__(self, def __init__(
bus: smbus.SMBus, self,
address: int = I2C_ADDRESS, bus: smbus.SMBus,
delay: int = 0, address: int = I2C_ADDRESS,
clear: bool = False): delay: int = 0,
clear: bool = False,
):
"""Initialize a ClusterCTRL/ClusterHAT driver instance for a given bus device.""" """Initialize a ClusterCTRL/ClusterHAT driver instance for a given bus device."""
self._bus = bus self._bus = bus
self._address = address self._address = address
@ -169,7 +173,9 @@ class ClusterCTRLv2Driver(object):
if (version := self._read(Reg.VERSION)) != 2: if (version := self._read(Reg.VERSION)) != 2:
raise IOError(f"Unsupported register format {version}; expected 2") raise IOError(f"Unsupported register format {version}; expected 2")
except: except:
raise ValueError("Cannot communicate with a ClusterCTRL/ClusterHAT on the given bus") raise ValueError(
"Cannot communicate with a ClusterCTRL/ClusterHAT on the given bus"
)
self._post_init() self._post_init()
@ -212,7 +218,7 @@ class ClusterCTRLv2Driver(object):
return self._bus.write_byte_data(self._address, id.value, val) return self._bus.write_byte_data(self._address, id.value, val)
def _call(self, op: Cmd, *args, clear = False): def _call(self, op: Cmd, *args, clear=False):
"""A convenient abstraction over the 'calling' convention for ops. """A convenient abstraction over the 'calling' convention for ops.
Operations are "called" when Reg.CMD is written to. Operations are "called" when Reg.CMD is written to.
@ -228,7 +234,19 @@ class ClusterCTRLv2Driver(object):
if self._clear or clear: if self._clear or clear:
args = chain(args, repeat(0)) args = chain(args, repeat(0))
args = zip([Reg.DATA0, Reg.DATA1, Reg.DATA2, Reg.DATA3, Reg.DATA4, Reg.DATA5, Reg.DATA6, Reg.DATA7], args) args = zip(
[
Reg.DATA0,
Reg.DATA1,
Reg.DATA2,
Reg.DATA3,
Reg.DATA4,
Reg.DATA5,
Reg.DATA6,
Reg.DATA7,
],
args,
)
for r, v in args: for r, v in args:
self._write(r, v) self._write(r, v)
@ -259,7 +277,7 @@ class ClusterCTRLv2Driver(object):
maxpi = self._read(Reg.MAXPI) maxpi = self._read(Reg.MAXPI)
if 0 <= id <= maxpi: if 0 <= id <= maxpi:
return id return id
else: else:
raise ValueError("Expected an id in [0,{maxpi:d}], got {id:d}") raise ValueError("Expected an id in [0,{maxpi:d}], got {id:d}")
@ -421,7 +439,7 @@ class ClusterCTRLv2Driver(object):
# 'order' (board ID) management # 'order' (board ID) management
#################################################################################################### ####################################################################################################
def get_order(self): def get_order(self):
"""Get the 'order' value of this device. Can be updated via """ """Get the 'order' value of this device. Can be updated via"""
return self._read(Reg.ORDER) return self._read(Reg.ORDER)

114
projects/clusterctrl/test/python/test_driver.py
View file

@ -90,7 +90,10 @@ def assert_log(bus, log):
log = [[simplify(e) for e in cmd] for cmd in log] log = [[simplify(e) for e in cmd] for cmd in log]
assert sublist(bus._log, log), "\n".join( assert sublist(bus._log, log), "\n".join(
["Failed to find expected sublog", "log:"] + [f"- {e}" for e in bus._log] + ["expected:"] + [f"- {e}" for e in log] ["Failed to find expected sublog", "log:"]
+ [f"- {e}" for e in bus._log]
+ ["expected:"]
+ [f"- {e}" for e in log]
) )
@ -98,103 +101,93 @@ def test_get_order(bus, driver):
"""Check that get_order sends the appropriate command sequence.""" """Check that get_order sends the appropriate command sequence."""
assert driver.get_order() == 13 assert driver.get_order() == 13
assert_log(bus, assert_log(bus, [["read", 0x20, Reg.ORDER]])
[["read", 0x20, Reg.ORDER]])
def test_set_order(bus, driver): def test_set_order(bus, driver):
"""Check that set_order sends the appropriate command sequence.""" """Check that set_order sends the appropriate command sequence."""
driver.set_order(14) driver.set_order(14)
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, 14], bus, [["write", 0x20, Reg.DATA0, 14], ["write", 0x20, Reg.CMD, Cmd.SET_ORDER]]
["write", 0x20, Reg.CMD, Cmd.SET_ORDER]]) )
def test_type(bus, driver): def test_type(bus, driver):
assert isinstance(driver.type, BoardType) assert isinstance(driver.type, BoardType)
assert_log(bus, assert_log(bus, [["read", 0x20, Reg.TYPE]])
[["read", 0x20, Reg.TYPE]])
def test_version(bus, driver): def test_version(bus, driver):
assert driver.fw_version assert driver.fw_version
# Invoke "read version" # Invoke "read version"
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, Data.VERSION], bus,
["write", 0x20, Reg.CMD, Cmd.GET_DATA]]) [
["write", 0x20, Reg.DATA0, Data.VERSION],
["write", 0x20, Reg.CMD, Cmd.GET_DATA],
],
)
# Read the two relevant registers # Read the two relevant registers
assert_log(bus, assert_log(bus, [["read", 0x20, Reg.DATA1], ["read", 0x20, Reg.DATA0]])
[["read", 0x20, Reg.DATA1],
["read", 0x20, Reg.DATA0]])
def test_reset(bus, driver): def test_reset(bus, driver):
driver.reset_all() driver.reset_all()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.RESET]])
[["write", 0x20, Reg.CMD, Cmd.RESET]])
def test_eeprom_save_all(bus, driver): def test_eeprom_save_all(bus, driver):
driver.eeprom_save_all() driver.eeprom_save_all()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE]])
[["write", 0x20, Reg.CMD, Cmd.SAVE]])
def test_eeprom_reset(bus, driver): def test_eeprom_reset(bus, driver):
driver.eeprom_reset() driver.eeprom_reset()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE_DEFAULTS]])
[["write", 0x20, Reg.CMD, Cmd.SAVE_DEFAULTS]])
def test_eeprom_save_powerstate(bus, driver): def test_eeprom_save_powerstate(bus, driver):
driver.eeprom_save_powerstate() driver.eeprom_save_powerstate()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE_POS]])
[["write", 0x20, Reg.CMD, Cmd.SAVE_POS]])
def test_eeprom_save_leds(bus, driver): def test_eeprom_save_leds(bus, driver):
driver.eeprom_save_leds() driver.eeprom_save_leds()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE_LEDS]])
[["write", 0x20, Reg.CMD, Cmd.SAVE_LEDS]])
def test_eeprom_save_order(bus, driver): def test_eeprom_save_order(bus, driver):
driver.eeprom_save_order() driver.eeprom_save_order()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE_ORDER]])
[["write", 0x20, Reg.CMD, Cmd.SAVE_ORDER]])
def test_eeprom_save_ussbboot(bus, driver): def test_eeprom_save_ussbboot(bus, driver):
driver.eeprom_save_usbboot() driver.eeprom_save_usbboot()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.SAVE_USBBOOT]])
[["write", 0x20, Reg.CMD, Cmd.SAVE_USBBOOT]])
def test_hub_on(bus, driver): def test_hub_on(bus, driver):
driver.hub_on() driver.hub_on()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.HUB_ON]])
[["write", 0x20, Reg.CMD, Cmd.HUB_ON]])
def test_hub_off(bus, driver): def test_hub_off(bus, driver):
driver.hub_off() driver.hub_off()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.HUB_OFF]])
[["write", 0x20, Reg.CMD, Cmd.HUB_OFF]])
def test_hub_reset(bus, driver): def test_hub_reset(bus, driver):
driver.hub_reset() driver.hub_reset()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.HUB_CYCLE]])
[["write", 0x20, Reg.CMD, Cmd.HUB_CYCLE]])
def test_power_status(bus, driver): def test_power_status(bus, driver):
bus._retvals.append(1) # Set a mocked return code. bus._retvals.append(1) # Set a mocked return code.
assert driver.power_status(1) == 1 assert driver.power_status(1) == 1
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, 1], bus, [["write", 0x20, Reg.DATA0, 1], ["write", 0x20, Reg.CMD, Cmd.GET_PSTATUS]]
["write", 0x20, Reg.CMD, Cmd.GET_PSTATUS]]) )
def test_pis(bus, driver): def test_pis(bus, driver):
@ -205,61 +198,60 @@ def test_power_all_on(bus, driver):
driver.power_all_on() driver.power_all_on()
for pi in driver.pis(): for pi in driver.pis():
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, pi.pi_id], bus,
["write", 0x20, Reg.CMD, Cmd.ON]]) [["write", 0x20, Reg.DATA0, pi.pi_id], ["write", 0x20, Reg.CMD, Cmd.ON]],
)
def test_power_all_off(bus, driver): def test_power_all_off(bus, driver):
driver.power_all_off() driver.power_all_off()
for pi in driver.pis(): for pi in driver.pis():
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, pi.pi_id], bus,
["write", 0x20, Reg.CMD, Cmd.OFF]]) [["write", 0x20, Reg.DATA0, pi.pi_id], ["write", 0x20, Reg.CMD, Cmd.OFF]],
)
def test_alert_on(bus, driver): def test_alert_on(bus, driver):
driver.alert_on() driver.alert_on()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.ALERT_ON]])
[["write", 0x20, Reg.CMD, Cmd.ALERT_ON]])
def test_alert_off(bus, driver): def test_alert_off(bus, driver):
driver.alert_off() driver.alert_off()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.CMD, Cmd.ALERT_OFF]])
[["write", 0x20, Reg.CMD, Cmd.ALERT_OFF]])
def test_fan_on(bus, driver): def test_fan_on(bus, driver):
driver.fan_on() driver.fan_on()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.DATA0, 1], ["write", 0x20, Reg.CMD, Cmd.FAN]])
[["write", 0x20, Reg.DATA0, 1],
["write", 0x20, Reg.CMD, Cmd.FAN]])
def test_fan_off(bus, driver): def test_fan_off(bus, driver):
driver.fan_off() driver.fan_off()
assert_log(bus, assert_log(bus, [["write", 0x20, Reg.DATA0, 0], ["write", 0x20, Reg.CMD, Cmd.FAN]])
[["write", 0x20, Reg.DATA0, 0],
["write", 0x20, Reg.CMD, Cmd.FAN]])
def test_fan_status(bus, driver): def test_fan_status(bus, driver):
driver.fan_status() driver.fan_status()
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, Data.FANSTATUS], bus,
["write", 0x20, Reg.CMD, Cmd.GET_DATA]]) [
["write", 0x20, Reg.DATA0, Data.FANSTATUS],
["write", 0x20, Reg.CMD, Cmd.GET_DATA],
],
)
def test_get_order(bus, driver): def test_get_order(bus, driver):
driver.get_order() driver.get_order()
assert_log(bus, assert_log(bus, [["read", 0x20, Reg.ORDER]])
[["read", 0x20, Reg.ORDER]])
def test_set_order(bus, driver): def test_set_order(bus, driver):
driver.set_order(253) driver.set_order(253)
assert_log(bus, assert_log(
[["write", 0x20, Reg.DATA0, 253], bus, [["write", 0x20, Reg.DATA0, 253], ["write", 0x20, Reg.CMD, Cmd.SET_ORDER]]
["write", 0x20, Reg.CMD, Cmd.SET_ORDER]]) )