Transports

Transport implementations Kv4pRadio can be constructed with.

class kv4p.transports.Kv4pTransport[source]

Bases: object

Documents the interface Kv4pRadio needs from a transport.

Not enforced via abc/@abstractmethod on purpose: the shape a future Bluetooth transport needs isn’t known yet, and a formal contract would just be a breaking change waiting to happen. Subclass and implement.

open(on_frame, on_error=None)[source]

Open the transport and start delivering incoming KISS frames to on_frame.

on_error is called (from the transport’s own background thread, if it has one) when the connection is lost unexpectedly, e.g. a disconnected USB cable or a device reboot mid-read. It is not called on a normal close().

Parameters:
  • on_frame (Callable[[int, bytes], None])

  • on_error (Callable[[Exception], None] | None)

Return type:

None

close()[source]

Close the transport.

Return type:

None

write_frame(command, payload)[source]

Write one KISS frame.

Parameters:
  • command (int)

  • payload (bytes)

Return type:

None

flush()[source]

Wait until pending output is written.

Return type:

None

reset()[source]

Reset the radio so it re-announces itself via HELLO.

Return type:

None

Serial Transport

class kv4p.transports.serial.Kv4pSerialTransport(device, baudrate)[source]

Bases: Kv4pTransport

Blocking serial transport with an RX thread, for use with Kv4pRadio.

Parameters:
  • device (str)

  • baudrate (int)

__init__(device, baudrate)[source]
Parameters:
  • device (str)

  • baudrate (int)

Return type:

None

open(on_frame, on_error=None)[source]

Open serial port and start the RX thread.

Parameters:
  • on_frame (Callable[[int, bytes], None])

  • on_error (Callable[[Exception], None] | None)

Return type:

None

close()[source]

Stop RX thread and close serial port.

Return type:

None

reset()[source]

Hardware-reset the ESP32 into its normal firmware via an RTS pulse.

DTR is held deasserted throughout so GPIO0 stays in run mode (not bootloader mode); RTS is pulsed to toggle EN/CHIP_PU.

Return type:

None

write_frame(command, payload)[source]

Write one KISS frame.

Parameters:
  • command (int)

  • payload (bytes)

Return type:

None

flush()[source]

Wait until serial output is written.

Return type:

None

Dummy Transport

class kv4p.transports.dummy.DummyTransport(*, firmware_version=17, window_size=2048, min_radio_freq=134.0, max_radio_freq=174.0, features=0, device_state_flags=0)[source]

Bases: Kv4pTransport

Transport with no real radio: answers reset() with a synthetic HELLO.

Parameters:
  • firmware_version (int)

  • window_size (int)

  • min_radio_freq (float)

  • max_radio_freq (float)

  • features (int)

  • device_state_flags (int)

__init__(*, firmware_version=17, window_size=2048, min_radio_freq=134.0, max_radio_freq=174.0, features=0, device_state_flags=0)[source]
Parameters:
  • firmware_version (int)

  • window_size (int)

  • min_radio_freq (float)

  • max_radio_freq (float)

  • features (int)

  • device_state_flags (int)

Return type:

None

open(on_frame, on_error=None)[source]

Open the transport and start delivering incoming KISS frames to on_frame.

on_error is called (from the transport’s own background thread, if it has one) when the connection is lost unexpectedly, e.g. a disconnected USB cable or a device reboot mid-read. It is not called on a normal close().

Parameters:
  • on_frame (Callable[[int, bytes], None])

  • on_error (Callable[[Exception], None] | None)

Return type:

None

close()[source]

Close the transport.

Return type:

None

reset()[source]

Reset the radio so it re-announces itself via HELLO.

Return type:

None

write_frame(command, payload)[source]

Write one KISS frame.

Parameters:
  • command (int)

  • payload (bytes)

Return type:

None

flush()[source]

Wait until pending output is written.

Return type:

None