diff --git a/README.md b/README.md index c4a7501..520e109 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,8 @@ __pizigate__ does require some additional adjustements on Raspberry Pi 3/Zero, a - [Raspberry Pi 3 and Raspberry Pi Zero configuration adjustements](https://zigate.fr/documentation/compatibilite-raspberry-pi-3-et-zero-w/) - [Raspberry Pi 4 configuration adjustements](https://zigate.fr/documentation/compatibilite-raspberry-pi-4-b/) +zigpy-zigate switches the __pizigate__ between running and flashing mode with GPIO 17 and 27, through `/dev/gpiomem`. This works on the Raspberry Pi 1 to 4 (including Zero and Compute Module 1 to 4) and needs read/write access to `/dev/gpiomem` (e.g. the `gpio` group on Raspberry Pi OS, or passing `/dev/gpiomem` through to a container). The Raspberry Pi 5 is not supported. + ## Flasher (ZiGate Firmware Tool) zigpy-zigate has an integrated Python "flasher" tool to flash firmware updates on your ZiGate (NXP Jennic JN5168). diff --git a/pyproject.toml b/pyproject.toml index c7fd9af..0df0e0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,6 @@ dependencies = [ "voluptuous", "zigpy>=1.3.0", "pyusb>=1.1.0", - "gpiozero", ] [tool.setuptools.packages.find] diff --git a/tests/test_gpio.py b/tests/test_gpio.py new file mode 100644 index 0000000..1f78371 --- /dev/null +++ b/tests/test_gpio.py @@ -0,0 +1,207 @@ +from unittest.mock import MagicMock + +import pytest + +from zigpy_zigate import gpio + + +@pytest.fixture +def board(tmp_path, monkeypatch): + """Point the board revision sources at temporary files, both missing initially.""" + device_tree = tmp_path / "linux,revision" + cpuinfo = tmp_path / "cpuinfo" + monkeypatch.setattr(gpio, "DEVICE_TREE_REVISION", str(device_tree)) + monkeypatch.setattr(gpio, "CPUINFO", str(cpuinfo)) + return device_tree, cpuinfo + + +@pytest.mark.parametrize( + "revision", + [ + 0x0002, # Pi 1 B, old-style + 0x0015, # Pi 1 A+, old-style + 0x1000010, # Pi 1 B+, old-style with the warranty bit set + 0x900093, # Zero 1.3 (BCM2835) + 0xA21041, # Pi 2 B (BCM2836) + 0xA02082, # Pi 3 B (BCM2837) + 0x902120, # Zero 2 W (BCM2837) + 0xB03114, # Pi 4 B (BCM2711) + 0xA03140, # CM4 (BCM2711) + 0x2C03114, # Pi 4 B with the warranty bit set + ], +) +def test_supported_board_device_tree(board, revision): + device_tree, _ = board + device_tree.write_bytes(revision.to_bytes(4, "big")) + + assert gpio.read_board_revision() == revision + gpio.check_supported_board() + + +def test_supported_board_cpuinfo(board): + _, cpuinfo = board + cpuinfo.write_text( + "processor\t: 0\nHardware\t: BCM2835\nRevision\t: c03114\n" + "Serial\t\t: 100000001234abcd\nModel\t\t: Raspberry Pi 4 Model B Rev 1.4\n" + ) + + assert gpio.read_board_revision() == 0xC03114 + gpio.check_supported_board() + + +def test_device_tree_preferred_over_cpuinfo(board): + device_tree, cpuinfo = board + device_tree.write_bytes((0xB03114).to_bytes(4, "big")) + cpuinfo.write_text("Revision\t: d04170\n") + + assert gpio.read_board_revision() == 0xB03114 + + +def test_truncated_device_tree_falls_back_to_cpuinfo(board): + device_tree, cpuinfo = board + device_tree.write_bytes(b"\x00\xb0") + cpuinfo.write_text("Revision\t: b03114\n") + + assert gpio.read_board_revision() == 0xB03114 + + +@pytest.mark.parametrize("revision", [0xC04170, 0xD04170]) # Pi 5 4GB, 8GB +def test_pi5_unsupported(board, revision): + device_tree, _ = board + device_tree.write_bytes(revision.to_bytes(4, "big")) + + with pytest.raises(gpio.GPIOError, match="BCM2712"): + gpio.check_supported_board() + + +def test_unknown_soc(board): + device_tree, _ = board + device_tree.write_bytes((0xA05170).to_bytes(4, "big")) + + with pytest.raises(gpio.GPIOError, match="SoC 5 "): + gpio.check_supported_board() + + +@pytest.mark.parametrize("revision", [0x0000, 0x0001, 0x0016]) +def test_unknown_old_style_revision(board, revision): + _, cpuinfo = board + cpuinfo.write_text(f"Revision\t: {revision:04x}\n") + + with pytest.raises(gpio.GPIOError, match="Unknown Raspberry Pi board revision"): + gpio.check_supported_board() + + +@pytest.mark.parametrize( + "cpuinfo_text", + [ + None, # no /proc/cpuinfo + "processor\t: 0\nBogoMIPS\t: 48.00\n", # e.g. x86 or a mainline arm64 kernel + "Revision\t: not-hex\n", + ], +) +def test_not_a_raspberry_pi(board, cpuinfo_text): + _, cpuinfo = board + if cpuinfo_text is not None: + cpuinfo.write_text(cpuinfo_text) + + assert gpio.read_board_revision() is None + with pytest.raises(gpio.GPIOError, match="Not a Raspberry Pi"): + gpio.check_supported_board() + + +def test_board_checked_before_opening_gpiomem(board, tmp_path, monkeypatch): + gpiomem = tmp_path / "gpiomem" + gpiomem.write_bytes(bytes(4096)) + monkeypatch.setattr(gpio, "GPIOMEM", str(gpiomem)) + os_open = MagicMock(side_effect=AssertionError("/dev/gpiomem opened")) + monkeypatch.setattr(gpio.os, "open", os_open) + + with pytest.raises(gpio.GPIOError, match="Not a Raspberry Pi"): + with gpio.GPIOMemory(): + pass + + os_open.assert_not_called() + + +def test_missing_gpiomem(board, tmp_path, monkeypatch): + device_tree, _ = board + device_tree.write_bytes((0xB03114).to_bytes(4, "big")) + monkeypatch.setattr(gpio, "GPIOMEM", str(tmp_path / "gpiomem")) + + with pytest.raises(gpio.GPIOError, match="Cannot open .*gpiomem"): + with gpio.GPIOMemory(): + pass + + +REG_GPFSEL1 = 0x04 // 4 # GPIO 10-19 +REG_GPFSEL2 = 0x08 // 4 # GPIO 20-29 +REG_GPSET0 = 0x1C // 4 +REG_GPCLR0 = 0x28 // 4 + + +@pytest.fixture +def gpiomem(board, tmp_path, monkeypatch): + """A Raspberry Pi 4 with a file standing in for `/dev/gpiomem`.""" + device_tree, _ = board + device_tree.write_bytes((0xB03114).to_bytes(4, "big")) + + path = tmp_path / "gpiomem" + path.write_bytes(bytes(4096)) + monkeypatch.setattr(gpio, "GPIOMEM", str(path)) + return path + + +def read_regs(path): + return memoryview(path.read_bytes()).cast("I") + + +def write_reg(path, index, value): + data = bytearray(path.read_bytes()) + regs = memoryview(data).cast("I") + regs[index] = value + regs.release() + path.write_bytes(data) + + +def test_set_output_only_changes_its_pin(gpiomem): + # Every pin in both registers set to ALT3 (0b111) + write_reg(gpiomem, REG_GPFSEL1, 0x3FFFFFFF) + write_reg(gpiomem, REG_GPFSEL2, 0x3FFFFFFF) + + with gpio.GPIOMemory() as mem: + mem.set_output(17) + mem.set_output(27) + + regs = read_regs(gpiomem) + assert regs[REG_GPFSEL1] == 0x3FFFFFFF & ~(0b110 << 21) + assert regs[REG_GPFSEL2] == 0x3FFFFFFF & ~(0b110 << 21) + + +def test_set_output_from_input(gpiomem): + with gpio.GPIOMemory() as mem: + mem.set_output(17) + mem.set_output(27) + + regs = read_regs(gpiomem) + assert regs[REG_GPFSEL1] == 0b001 << 21 + assert regs[REG_GPFSEL2] == 0b001 << 21 + + +def test_write(gpiomem): + with gpio.GPIOMemory() as mem: + mem.write(17, True) + assert read_regs(gpiomem)[REG_GPSET0] == 1 << 17 + + mem.write(27, False) + assert read_regs(gpiomem)[REG_GPCLR0] == 1 << 27 + + regs = read_regs(gpiomem) + assert regs[REG_GPSET0] == 1 << 17 + assert regs[REG_GPCLR0] == 1 << 27 + + +def test_reopen(gpiomem): + # The mapping is released on exit, so it can be opened again (e.g. on reconnect) + for _ in range(2): + with gpio.GPIOMemory() as mem: + mem.write(17, True) diff --git a/tests/test_uart.py b/tests/test_uart.py index 4f9740b..694bb0b 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -1,6 +1,5 @@ from unittest.mock import AsyncMock, MagicMock, call -import gpiozero import pytest from serialx import SerialPortInfo import zigpy.config @@ -21,7 +20,8 @@ def gw(): ("/dev/null", "pizigate:/dev/ttyAMA0"), ) async def test_connect(port, monkeypatch): - monkeypatch.setattr(gpiozero.Device, "_default_pin_factory", MagicMock()) + monkeypatch.setattr(common, "GPIOMemory", MagicMock()) + monkeypatch.setattr(common.time, "sleep", MagicMock()) api = MagicMock() @@ -186,8 +186,44 @@ def test_is_not_zigate_wifi(): assert r is False -def test_startup_gpio_toggling(monkeypatch): - monkeypatch.setattr(gpiozero.Device, "_default_pin_factory", MagicMock()) +def mock_gpio(monkeypatch): + calls = MagicMock() + gpio_memory = MagicMock() + gpio_memory.return_value.__enter__.return_value = calls.gpio + monkeypatch.setattr(common, "GPIOMemory", gpio_memory) + monkeypatch.setattr(common.time, "sleep", calls.sleep) + return calls + + +def test_pizigate_running_mode(monkeypatch): + calls = mock_gpio(monkeypatch) common.set_pizigate_running_mode() + + assert calls.mock_calls == [ + call.gpio.set_output(17), + call.gpio.set_output(27), + call.gpio.write(27, True), + call.sleep(0.5), + call.gpio.write(17, False), + call.sleep(0.5), + call.gpio.write(17, True), + call.sleep(0.5), + ] + + +def test_pizigate_flashing_mode(monkeypatch): + calls = mock_gpio(monkeypatch) + common.set_pizigate_flashing_mode() + + assert calls.mock_calls == [ + call.gpio.set_output(17), + call.gpio.set_output(27), + call.gpio.write(27, False), + call.sleep(0.5), + call.gpio.write(17, False), + call.sleep(0.5), + call.gpio.write(17, True), + call.sleep(0.5), + ] diff --git a/zigpy_zigate/common.py b/zigpy_zigate/common.py index 3469c95..8c6dbf7 100644 --- a/zigpy_zigate/common.py +++ b/zigpy_zigate/common.py @@ -4,34 +4,16 @@ import re import time -from gpiozero import OutputDevice from serialx import SerialException, list_serial_ports +from zigpy_zigate.gpio import GPIOMemory + LOGGER = logging.getLogger(__name__) GPIO_PIN0 = 17 GPIO_PIN2 = 27 -class UnclosableOutputDevice(OutputDevice): - """ - `OutputDevice` that never closes its pins. Allows for the last-written pin state to - be retained even after the `OutputDevice` is garbage collected. - """ - - def __init__( - self, pin=None, *, active_high=True, initial_value=False, pin_factory=None - ): - super().__init__( - pin, - active_high=active_high, - initial_value=initial_value, - pin_factory=pin_factory, - ) - self._pin.close = lambda *args, **kwargs: None - self.pin_factory.close = lambda *args, **kwargs: None - - def discover_port(): """discover zigate port""" ports = list_serial_ports() @@ -78,33 +60,35 @@ def is_zigate_wifi(port): def set_pizigate_running_mode(): LOGGER.info("Put PiZiGate in running mode") - gpio0 = UnclosableOutputDevice(pin=GPIO_PIN0, initial_value=None) - gpio2 = UnclosableOutputDevice(pin=GPIO_PIN2, initial_value=None) + with GPIOMemory() as gpio: + gpio.set_output(GPIO_PIN0) + gpio.set_output(GPIO_PIN2) - gpio2.on() - time.sleep(0.5) + gpio.write(GPIO_PIN2, True) + time.sleep(0.5) - gpio0.off() - time.sleep(0.5) + gpio.write(GPIO_PIN0, False) + time.sleep(0.5) - gpio0.on() - time.sleep(0.5) + gpio.write(GPIO_PIN0, True) + time.sleep(0.5) def set_pizigate_flashing_mode(): LOGGER.info("Put PiZiGate in flashing mode") - gpio0 = UnclosableOutputDevice(pin=GPIO_PIN0, initial_value=None) - gpio2 = UnclosableOutputDevice(pin=GPIO_PIN2, initial_value=None) + with GPIOMemory() as gpio: + gpio.set_output(GPIO_PIN0) + gpio.set_output(GPIO_PIN2) - gpio2.off() - time.sleep(0.5) + gpio.write(GPIO_PIN2, False) + time.sleep(0.5) - gpio0.off() - time.sleep(0.5) + gpio.write(GPIO_PIN0, False) + time.sleep(0.5) - gpio0.on() - time.sleep(0.5) + gpio.write(GPIO_PIN0, True) + time.sleep(0.5) def ftdi_set_bitmode(dev, bitmask): diff --git a/zigpy_zigate/gpio.py b/zigpy_zigate/gpio.py new file mode 100644 index 0000000..7dfe2a9 --- /dev/null +++ b/zigpy_zigate/gpio.py @@ -0,0 +1,116 @@ +"""Minimal Raspberry Pi GPIO access through `/dev/gpiomem`.""" + +import mmap +import os + +GPIOMEM = "/dev/gpiomem" +DEVICE_TREE_REVISION = "/proc/device-tree/system/linux,revision" +CPUINFO = "/proc/cpuinfo" + +# SoC field (bits 12-15) of new-style board revision codes +SOC_NAMES = {0: "BCM2835", 1: "BCM2836", 2: "BCM2837", 3: "BCM2711", 4: "BCM2712"} + +# Raspberry Pi 1-4, Zero and CM1-4: their GPIO block has the BCM2835 register layout. +# The Pi 5 (BCM2712) has its header GPIOs on the RP1 chip, with different registers. +SUPPORTED_SOCS = {0, 1, 2, 3} + + +class GPIOError(RuntimeError): + """The PiZiGate's GPIO pins cannot be driven.""" + + +def read_board_revision() -> int | None: + """Read the Raspberry Pi board revision code, from the same places as gpiozero.""" + try: + with open(DEVICE_TREE_REVISION, "rb") as f: + data = f.read() + if len(data) == 4: + return int.from_bytes(data, "big") + except OSError: + pass + + try: + with open(CPUINFO) as f: + for line in f: + if line.startswith("Revision"): + return int(line.partition(":")[2], 16) + except (OSError, ValueError): + pass + + return None + + +def check_supported_board() -> None: + """Raise `GPIOError` unless this is a Raspberry Pi with BCM2835-style GPIO.""" + revision = read_board_revision() + if revision is None: + raise GPIOError( + "Not a Raspberry Pi: no board revision code in" + f" {DEVICE_TREE_REVISION} or {CPUINFO}" + ) + + if revision & 0x800000: + # New-style revision code + soc = (revision >> 12) & 0xF + elif 0x0002 <= revision & 0xFFFFFF <= 0x0015: + # Old-style revision codes are all Raspberry Pi 1 models + soc = 0 + else: + raise GPIOError(f"Unknown Raspberry Pi board revision code {revision:#x}") + + if soc not in SUPPORTED_SOCS: + raise GPIOError( + f"Unsupported Raspberry Pi SoC {SOC_NAMES.get(soc, soc)} (board revision" + f" {revision:#x}): PiZiGate GPIO control supports the Raspberry Pi 1-4" + ) + + +class GPIOMemory: + # Register byte offsets, from the BCM2835 ARM Peripherals datasheet, section 6.1 + GPFSEL0 = 0x00 # function select: 3 bits per pin, 10 pins per register + GPSET0 = 0x1C # write 1 to drive a pin high: 1 bit per pin, 32 pins per register + GPCLR0 = 0x28 # write 1 to drive a pin low: 1 bit per pin, 32 pins per register + + FUNCTION_WIDTH = 3 + FUNCTION_MASK = 0b111 + FUNCTION_OUTPUT = 0b001 + + def __enter__(self) -> "GPIOMemory": + check_supported_board() + + try: + fd = os.open(GPIOMEM, os.O_RDWR | os.O_SYNC) + except OSError as exc: + raise GPIOError(f"Cannot open {GPIOMEM}: {exc.strerror}") from exc + + try: + self._mmap = mmap.mmap(fd, 4096) + finally: + os.close(fd) + self._regs = memoryview(self._mmap).cast("I") # native 32-bit register access + return self + + def __exit__(self, *exc_info) -> None: + self._regs.release() + self._mmap.close() + + def _index(self, base: int, register: int) -> int: + return base // 4 + register + + def set_output(self, pin: int) -> None: + register, field = divmod(pin, 10) + shift = field * self.FUNCTION_WIDTH + index = self._index(self.GPFSEL0, register) + + # Other pins share this register, so only replace this pin's field + value = self._regs[index] + value &= ~(self.FUNCTION_MASK << shift) + value |= self.FUNCTION_OUTPUT << shift + self._regs[index] = value + + def write(self, pin: int, value: bool) -> None: + register, bit = divmod(pin, 32) + base = self.GPSET0 if value else self.GPCLR0 + + # Zero bits are ignored, so this only affects `pin` + self._regs[self._index(base, register)] = 1 << bit