diff --git a/README.md b/README.md index aa90bda..51916de 100644 --- a/README.md +++ b/README.md @@ -154,9 +154,17 @@ on: 1, 3 [0x05] Probing never writes to a bit-banged board. Its FTDI chip is a FIFO rather than a UART, so every byte written to it lands straight on the relays — asking it for its -state would leave them holding a `/`. It gives itself away by handing over bytes with -nothing asked of it, which a board that answers a protocol never does, so the port is -listened to before it is spoken to, and relays are left where they were. +state would leave them holding a `/`, and the board would then answer with enough +bytes to pass for a 16 relay board that is not there. + +So a bit-banged board is ruled out first, without writing anything. Reading the chip's +data lines changes nothing — no byte is sent and no pin is switched to an output — but +it wakes the read side of the FIFO, and the board then hands the port bytes with +nothing asked of it, which a board that answers a protocol does not do. Relays are +left exactly where they were. + +This needs `libftdi`. Without it nothing can be told about the data lines and the +board is left to the protocol probe, which is where `--board` comes in. ## Board support diff --git a/src/denkovi_cli/bitbang.py b/src/denkovi_cli/bitbang.py index 876ff10..fd4ca31 100644 --- a/src/denkovi_cli/bitbang.py +++ b/src/denkovi_cli/bitbang.py @@ -122,10 +122,45 @@ def _connected(self) -> Any: return self._device -def _pylibftdi() -> tuple[Any, Any, type[BaseException]]: - """Return the pylibftdi names used here, or say how to install it.""" +def read_data_lines(serial_number: str | None) -> int: + """Return the state of the eight data lines of a board's FTDI chip. + + The chip is opened in its ordinary serial mode and only read from: no byte + is written and no pin is switched to an output, so this is safe to do to a + board of any kind. On a bit-banged board the data lines are the relays. + + Reading them also wakes the read side of an FT245's FIFO, which is what + lets a bit-banged board be recognised without writing to it; see + `board.probe_board_type`. + """ + from ctypes import byref, c_ubyte + + device_class, driver, ftdi_error = _pylibftdi(bit_bang=False) + try: + device = device_class( + serial_number or None, + driver=_driver(driver), + auto_detach=sys.platform != "darwin", + ) + try: + pins = c_ubyte() + if device.ftdi_fn.ftdi_read_pins(byref(pins)) != 0: + raise DenkoviError(f"could not read the data lines of {serial_number}.") + return int(pins.value) + finally: + device.close() + except (ftdi_error, OSError, AttributeError) as error: + raise DenkoviError(_open_failed(error, serial_number)) from error + + +def _pylibftdi(*, bit_bang: bool = True) -> tuple[Any, Any, type[BaseException]]: + """Return the pylibftdi names used here, or say how to install it. + + ``bit_bang`` picks the device class: the bit-bang one drives the data lines + as outputs, the plain one leaves the chip in the serial mode it was in. + """ try: - from pylibftdi import BitBangDevice, Driver, FtdiError + from pylibftdi import BitBangDevice, Device, Driver, FtdiError except ImportError as error: raise DenkoviError( "the 4 and 8 relay boards are driven through pylibftdi, which is not " @@ -133,7 +168,7 @@ def _pylibftdi() -> tuple[Any, Any, type[BaseException]]: "libftdi C library: 'brew install libftdi' on macOS, or the distribution's " "libftdi1 package on Linux)." ) from error - return BitBangDevice, Driver, FtdiError + return (BitBangDevice if bit_bang else Device), Driver, FtdiError def _driver(driver_class: Any) -> Any: diff --git a/src/denkovi_cli/board.py b/src/denkovi_cli/board.py index f758fb1..c65a8f2 100644 --- a/src/denkovi_cli/board.py +++ b/src/denkovi_cli/board.py @@ -183,53 +183,73 @@ def _describe(devices: list[Device]) -> str: ) -def probe_board_type(port: str, *, timeout: float = 1.0) -> str | None: - """Return the board type on ``port``, or ``None`` if it cannot be told. +def probe_board_type(device: Device, *, timeout: float = 1.0) -> str | None: + """Return the type of ``device``, or ``None`` if it cannot be told. Only the VCP boards can be identified over the wire: they answer the ``ask`` command with one status byte per eight relays. The bit-banged 4 and 8 relay boards cannot be told apart from each other, so they have to be named explicitly. - Nothing is written to a board that must not be written to. A bit-banged - board is a FIFO wearing a serial port's clothes: every byte written to it - lands on the relays, and 'ask//' would leave them holding a '/'. It gives - itself away by handing over bytes with nothing asked of it, which a board - that answers a protocol never does, so the port is listened to first and - only a port that stays quiet is spoken to. + A bit-banged board is ruled out before a single byte is written, because + writing to one is what must not happen: it is a FIFO wearing a serial + port's clothes, so 'ask//' would land on the relays and leave them holding + a '/'. Reading its data lines, which changes nothing, wakes the read side + of its FIFO, and it then hands the port bytes with nothing asked of it — + which a board that answers a protocol does not do. """ + if _looks_bit_banged(device): + return None + try: - with serial.Serial(port=port, baudrate=9600, timeout=LISTEN_TIMEOUT) as connection: + with serial.Serial(port=device.port, baudrate=9600, timeout=timeout) as connection: time.sleep(DEFAULT_DELAY) connection.reset_input_buffer() connection.reset_output_buffer() - if connection.read(1): - return None - connection.timeout = timeout connection.write(b"ask//") time.sleep(DEFAULT_DELAY) reply = connection.read(2) except (OSError, serial.SerialException) as error: - raise DenkoviError(f"could not open {port}: {error}") from error + raise DenkoviError(f"could not open {device.port}: {error}") from error if len(reply) == 2: return dae_RelayBoard.DAE_RELAYBOARD_TYPE_16 return None -def resolve_board_type(port: str, board_type: str | None) -> str: - """Return the board type to drive ``port`` with, probing when not given.""" +def _looks_bit_banged(device: Device) -> bool: + """Whether the board is one whose relays hang off its FTDI data lines. + + Answers ``False`` when it cannot tell — because libftdi is not installed, + or the chip cannot be opened — which leaves the board to the VCP probe, + where the worst case is a board that has to be named with ``--board``. + """ + try: + from .bitbang import is_supported, read_data_lines + + if not is_supported(): + return False + read_data_lines(device.serial_number) + with serial.Serial(port=device.port, baudrate=9600, timeout=LISTEN_TIMEOUT) as connection: + connection.reset_input_buffer() + return bool(connection.read(1)) + except (DenkoviError, OSError, serial.SerialException): + return False + + +def resolve_board_type(device: Device, board_type: str | None) -> str: + """Return the board type to drive ``device`` with, probing when not given.""" if board_type is not None: if board_type not in BOARD_TYPES: supported = ", ".join(BOARD_TYPES) raise DenkoviError(f"unknown board type {board_type!r}. Supported: {supported}.") return board_type - detected = probe_board_type(port) + detected = probe_board_type(device) if detected is None: raise DenkoviError( - f"could not identify the board on {port}. The 4 and 8 relay boards cannot " - "be detected over the wire; pass --board type4 or --board type8." + f"could not identify the board on {device.port}. The 4 and 8 relay boards " + "cannot be detected over the wire; pass --board type4 or --board type8." ) return detected diff --git a/src/denkovi_cli/cli.py b/src/denkovi_cli/cli.py index 68d05e7..4b4c6eb 100644 --- a/src/denkovi_cli/cli.py +++ b/src/denkovi_cli/cli.py @@ -285,7 +285,7 @@ def command_watch(args: argparse.Namespace) -> int: def _connect(args: argparse.Namespace): device = resolve_device(args.port, args.serial) - board_type = resolve_board_type(device.port, args.board) + board_type = resolve_board_type(device, args.board) return open_board(device, board_type, delay=args.delay) diff --git a/tests/test_bitbang.py b/tests/test_bitbang.py index d68c0ec..768533a 100644 --- a/tests/test_bitbang.py +++ b/tests/test_bitbang.py @@ -275,33 +275,61 @@ class TestProbe: """Probing must never write to a board whose relays are the data lines.""" @pytest.fixture - def port(self, monkeypatch: pytest.MonkeyPatch) -> type[FakePort]: + def port(self, monkeypatch: pytest.MonkeyPatch): def install(fake: FakePort) -> FakePort: monkeypatch.setattr(board_module.serial, "Serial", fake) return fake - return install # type: ignore[return-value] + return install - def test_a_talkative_port_is_left_alone(self, port) -> None: - # An FT245 hands over bytes with nothing asked of it, and anything - # written to it would land on the relays. - fake = port(FakePort(stream=b"\x00" * 8)) + @pytest.fixture + def data_lines(self, monkeypatch: pytest.MonkeyPatch): + """Make reading the FTDI data lines succeed, fail, or be unavailable.""" + + def install(result: object) -> None: + def read(serial_number: str | None) -> int: + if isinstance(result, Exception): + raise result + return int(result) # type: ignore[arg-type] + + monkeypatch.setattr(bitbang, "read_data_lines", read) + monkeypatch.setattr(bitbang, "is_supported", lambda: True) + + return install + + def test_a_board_whose_fifo_answers_is_never_written_to(self, port, data_lines) -> None: + # Reading the data lines wakes an FT245's FIFO, and it then hands over + # bytes unprompted. Anything written would have landed on the relays. + data_lines(0x5A) + fake = port(FakePort(stream=b"\x5a" * 4)) - assert board_module.probe_board_type("/dev/fake") is None + assert board_module.probe_board_type(_device()) is None assert fake.written == b"" - def test_a_board_that_answers_is_a_type16(self, port) -> None: + def test_a_board_that_answers_the_protocol_is_a_type16(self, port, data_lines) -> None: + data_lines(0x00) fake = port(FakePort(reply=b"\x00\x00")) - assert board_module.probe_board_type("/dev/fake") == TYPE16 + assert board_module.probe_board_type(_device()) == TYPE16 assert fake.written == b"ask//" - def test_a_board_that_says_nothing_at_all_is_unknown(self, port) -> None: + def test_a_board_that_says_nothing_at_all_is_unknown(self, port, data_lines) -> None: + data_lines(0x00) port(FakePort()) - assert board_module.probe_board_type("/dev/fake") is None + assert board_module.probe_board_type(_device()) is None + + def test_without_libftdi_the_protocol_probe_still_runs(self, port, data_lines) -> None: + # Nothing can be told about the data lines, so the board is left to the + # VCP probe, exactly as it was before there was a way to read them. + data_lines(DenkoviError("no libftdi here")) + fake = port(FakePort(reply=b"\x00\x00")) + + assert board_module.probe_board_type(_device()) == TYPE16 + assert fake.written == b"ask//" - def test_an_unidentified_board_asks_to_be_named(self, port) -> None: - port(FakePort(stream=b"\x00" * 8)) + def test_an_unidentified_board_asks_to_be_named(self, port, data_lines) -> None: + data_lines(0x5A) + port(FakePort(stream=b"\x5a" * 4)) with pytest.raises(DenkoviError, match="--board type8"): - board_module.resolve_board_type("/dev/fake", None) + board_module.resolve_board_type(_device(), None)