signal module

Base classes for translating GPIO signals into Linux kernel key events. Supported signals have fixed or variable pulse lengths or individual edges, with each signal received on one specific GPIO. Key events are exposed as instances of KeyEvent.

GPIO backend must implement PigpioMinimalApi and PigpioCallbackApi to work with this module.

Furthermore, custom protocol handlers can be created by extending the classes in this module. See topic for an entry point into the extensibility mechanism. Existing extensions are:

For an overview of how signals can be processed, the Configuration classes are a good starting point.

Configuration classes

The classes in this section evaluate and hold configurations for signal protocols, signal listeners and GPIO mappings. They can be easily initialized with dict instances, e.g. read from JSON files.

Auto-discovering and parsing JSON files are out of scope for gpiosvr.signal. See the ctlbase package instead. The following is a sample JSON representation of a signal configuration:

{
  "evdevName": "pi-sig-injector",
  "gpios": {
    "25": "door-bell"
  },
  "signalProtocols": {
    "door-bell": {
      "basePulse_μs": 9000,
      "basePulseCount": 5,
      "codeStartBit": 1,
      "preamble_ms": 10,
      "postamble_ms": 12,
      "tolerancePercentage": 30,
      "debounce_μs": 600,
      "signalListener": {
        "mayInject": false
      },
      "keys": {
        "0x1f": {
          "keyName": "KEY_SOUND"
        }
      }
    }
  }
}
class gpiosvr.signal.SignalConfig(signalConfig: dict, topic='signal', protocolClass_=<class 'gpiosvr.signal.SignalProtocolDescription'>, listenerConfigClass_=<class 'gpiosvr.signal.SignalListenerConfig'>)
__init__(signalConfig: dict, topic='signal', protocolClass_=<class 'gpiosvr.signal.SignalProtocolDescription'>, listenerConfigClass_=<class 'gpiosvr.signal.SignalListenerConfig'>)

Properties describing a compound configuration for signal processing on multiple GPIOs.

They combine the configurations of signal protocols (SignalProtocolDescription), signal listeners (SignalListenerConfig) and GPIO mappings.

Parameters:

  • signalConfig (dict) –

    Under the key evdevName (str), the dict must contain a str for the logical name of the event device to which key events will be published. This is not the name of the event device under /dev/input/. Instead, it is a logical name that can be referenced in the inputDeviceName positional parameter of the gpiosvr.bin.key_monitor executable.

    Under the key gpios (str), the dict must contain a dict mapping GPIO BCM numbers to protocol names. For instance, the GPIO number 0 (str) can be mapped to a protocol name mybutton. This protocol name must then be referenced in the following signal protocol descriptions.

    Under the key signalProtocols (str), the dict must contain a list of signal protocol descriptions. These in turn must be dict instances to be passed to fromConfig(). To map a protocol description to a specific GPIO, a name must be provided for each signal protocol.

    In a addition to a name, each signal protocol must feature a keys (str) key with another dict mapping hex codes to Linux key names. See gpiosvr.key.KeysConfig for reference.

    Optionally, each signal protocol may each feature a signalListener (str) key with another dict configuring the listener for that specific combination of GPIO and protocol. See __init__() for reference.

  • topic (str, optional) – For extensibility, an identifier for handling specific types of protocols. It is used as a prefix for configuration keys when parsing the configuration given in signalConfig. For instance, if set to rc, the keys expected in signalConfig are rcProtocols and rcListener. this allows for combining multiple configurations in one dict instance. Defaults to signal.

  • protocolClass_ (class object, optional) – For extensibility, the class to which parsing of the signal protocol descriptions is delegated. The class needs to implement a fromConfig method accepting a dict instance. Defaults to SignalProtocolDescription.

  • listenerConfigClass_ (class object, optional) – For extensibility, the class to which parsing of the signal listener configurations is delegated. The class needs to implement a fromConfig method accepting a dict instance. Defaults to SignalListenerConfig.

class gpiosvr.signal.SignalProtocolDescription(name=None, basePulse_μs=None, tolerancePercentage=None, minEdgeCount=None, basePulseCount=None, preamble_ms=None, postamble_ms=None, codeStartBit=-1, bitOrder=None, edgeEvaluationInterval_ms=None, debounce_μs=None)
__init__(name=None, basePulse_μs=None, tolerancePercentage=None, minEdgeCount=None, basePulseCount=None, preamble_ms=None, postamble_ms=None, codeStartBit=-1, bitOrder=None, edgeEvaluationInterval_ms=None, debounce_μs=None)

Parameters describing a signal protocol, i.e. the expected pulses or edges and how to translate them to hex codes. The hex codes can be mapped to Linux key names in the scope of SignalConfig.

Before translating a signal to a hex code, it is first translated to a sequence of logical bits. The translation algorithm can be extended by custom signal listeners.

In the basic SignalListener, the logical bit is simply alternating with the GPIO level. To set the first logical bit, use the parameter codeStartBit. The number of bits decoded from the signal, the so called bitCount, is derived from the parameters basePulse_μs, minEdgeCount and basePulseCount. See paramsToBitCount().

Parameters:

  • name (str, optional) – Identifying name of the protocol for mapping it to a specific GPIO. Also used in messages and asyncio task names. Defaults to None.

  • basePulse_μs (int, optional) – For a signal with a fixed pulse length, the base pulse length in microseconds. Defaults to 0.

  • tolerancePercentage (int, optional) – For a signal with a fixed pulse length, the maximum tolerated deviation of basePulse_μs (plus or minus). Specified as a percentage value from 0 to 50. Defaults to TOLERANCE_PERCENTAGE_DEFAULT.

  • minEdgeCount (int, optional) – For a signal with variable pulse lengths or for individual edges, the minimum number of expected edges that marks the end of the signal. Defaults to 0.

  • basePulseCount (int, optional) – For a signal with a fixed pulse length, the number of base pulses that marks the end of the signal. For instance, for a basePulse_μs of 45, and a basePulseCount of 10, a hex code will be produced after 45 * 10 = 450 microseconds. Defaults to None.

  • preamble_ms (int, optional) – The duration of “silence” in milliseconds, that marks the start of a signal on the first edge. For instance, if set to 150, a signal will be considered “started” on the first edge after at least 150 milliseconds without any other edge. Defaults to basePulse_μs * 1.5 // 1000.

  • postamble_ms (int, optional) – The duration of “silence” in milliseconds, that marks the end of the signal after the previous edge. For instance, if set to 300, the signal will be considered “ended” if no other edge has occurred for at least 300 milliseconds since the previous edge. The value must be larger than preamble_ms so that no new signal is considered “started” before the current signal is considered “ended”. Defaults to (bitCount + 0.5) * basePulse_μs // 1000.

  • codeStartBit (int | str, optional) – The first logical bit in the decoded signal. It may be 0, 1 for a fixed logical value. It may also be GPIO_LEVEL (str) for the actual GPIO level of the first edge that marks the start of the signal. Defaults to CODE_START_BIT_DEFAULT.

  • bitOrder (BitOrder) – The bit order to use when translating the logical bits of a signal to a hex code. For instance, if the signal is decoded to a logical bit sequence of 1001101, it will be translated to hex code 0x4d if the bit order is MSB_FIRST. By contrast, it will be translated to 0x59, if the bit order is LSB_FIRST. Defaults to BIT_ORDER_DEFAULT.

  • edgeEvaluationInterval_ms (int, optional) – The interval in milliseconds in which queued edges are evaluated as fixed or variable-length pulses or individual edges. For bursts of pulses, the value must be high enough to completely queue all edges before evaluating them. However, if too long, the decoding into logical bits and translating them to a hex code could be perceptibly delayed. Lower values result in higher CPU load.

  • debounce_μs (int, optional) – The time span in microseconds, within which two edges shall be ignored. Useful e.g. for hardware buttons that may produce high-frequency edges due to mechanical bouncing. May also be used to filter out high-frequency noise on a line. Defaults to DEBOUNCE_μS_DEFAULT.

Raises:

ValueError – if a parameter value is missing or is invalid, given the other parameter values.

static fromConfig(name, protocolConfig: dict)

Convenience method for creating a signal protocol description from a configuration dict.

The dict entries may only contain the parameter names and values of __init__().

Returns:

A signal protocol description representing the configuration passed in via protocolConfig.

Return type:

SignalProtocolDescription

Raises:

The same exceptions as with __init__().

class gpiosvr.signal.SignalListenerConfig(signaListenerConfig: dict)
__init__(signaListenerConfig: dict)

Properties describing a signal listener configuration.

Parameters:

signaListenerConfig (dict) –

Under the key mayInject (str), the dict may contain a bool value. If True, the signal listener will accept key events injected via the injectKeyCode() method. This is security-relevant, as key codes can then be injected by malicious software without the need to be physically attached to the corresponding GPIO.

Under the key isDebugOutput (str), the dict may contain a bool value. If True, debug messages will be written to the message buffer of the signal listener, if any. Defaults to None.

Under the key isTesting (str), the dict may contain a bool value. If True, more verbose messages about occurring edges will be written to the message buffer of the signal listener, if any. Useful for finding out the actual pulse or edge sequences on a line. Always implies isDebugOutput == True. Defaults to None.

Raises:

ValueError if isDebugOutput is set to False while isTesting is set to True

Pulse event classes

class gpiosvr.signal.PulseEvent(pulses_μs: list, pulseStartLevel, timestamp_μs)

Represents a completed pulse sequence.

class gpiosvr.signal.PulsesInterrupted(pulses_μs=(), pulseStartLevel=None, timestamp_μs=None)

Bases: PulseEvent, Exception

Represents an incompleted pulse sequence, i.e. some interruption to a pulse sequence.

class gpiosvr.signal.PulseEventIterator(loop, gpioClient, gpio, protocol=None, maxBasePulseCount=100, msgBuffer=None, isDebugOutput=False, isTesting=False)

An async iterator over pulse events, based on a signal protocol.

It is the main class receiving edge callbacks and evaluating them based on a signal protocol description. The iterator yields instances of PulseEvent that can be interpreted as signals, decoded into logical bits and translated to hex codes.

The iterator lazily sets up the edge callback with its gpioClient instance once the __anext__() method is called for the first time. Once started, the iterator runs indefinitely, until aclose() is called.

__init__(loop, gpioClient, gpio, protocol=None, maxBasePulseCount=100, msgBuffer=None, isDebugOutput=False, isTesting=False)
Parameters:
async onArtificialEdge()

Internal event callback. It is called when an outside requestor has triggered an artificial edge via provokeArtificialEdge().

The implementation is expected to put a corresponding PulseEvent instance into the event queue, then return True to show that the event has been handled.

async onPulseInterruption()

Internal event callback.. It is called when the pulse sequence has been interrupted, considering the protocol.

If the implementation wishes to discard the incomplete pulse sequence, it must return True. If the implementation wishes to append further pulses to the incomplete pulse sequence, it must return False. This is the default.

If the implementation whishes to notify coroutines iterating over PulseEvent instances, it must put an instance of PulsesInterrupted into the event queue, then return True.

async onSequenceEnd()

Internal event callback. It is called when the pulse sequence has completed, considering the protocol.

The implementation is expected to put a corresponding PulseEvent instance into the event queue, then return True to show that the event has been handled.

async onSingleEdge()

Internal event callback. It is called when the protocol aims at single edges and a corresponding edge has occurred.

The implementation is expected to put a corresponding PulseEvent instance into the event queue, then return True to show that the event has been handled.

class gpiosvr.signal.SignalListener(loop, gpioClient, gpio, mayInject=False, protocol=None, maxBasePulseCount=100, msgBuffer=None, isDebugOutput=False, isTesting=False)

Main class for listening to signals and producing key events.

Pulse events are decoded to signals with logical bits, then translated to hex codes, eventually mapped and exposed as key events.

__init__(loop, gpioClient, gpio, mayInject=False, protocol=None, maxBasePulseCount=100, msgBuffer=None, isDebugOutput=False, isTesting=False)
Parameters:
static fromConfig(loop, gpioClient, gpio, protocol, msgBuffer, listenerConfig)

Convenience method for creating a signal listener from a configuration.

Parameters:

listenerConfig (object) – Either a dict or an instance of SignalListenerConfig. If the former, the dict keys must be identical to the parameter names of __init__().

Returns:

A SignalListener instance representing the configuration passed in via listenerConfig.

Return type:

SignalListener

Raises:

The same exceptions as __init__().

createPulseEventIterator()

Returns the pulse event iterator to use in waitForKeyEvents().

May be overridden for extensibility.

Returns:

The pulse event iterator to use. May be a cached object in case waitForKeyEvents() is called multiple times.

Return type:

PulseEventIterator

async pulseEventToKeyEvents(pulseEventOrKeyCodes, debugOutput=None)

Translates pulse events to key events.

May be overridden for extensibility.

Parameters:

  • pulseEventOrKeyCodes (PulseEvent | PulsesInterrupted | list[str] | tuple[str]) – Either a pulse event to translate or, as a short cut, a sequence of desired key codes if they are already known. Key codes must be strings starting with 0x.

  • debugOutput (StringIO, optional) – A string buffer for appending debug messages. Defaults to None.

async waitForKeyEvents()

May be used in an async for loop to wait for key events translated from received signals.

async injectKeyCode(keyCode)

When waitForKeyEvents() is running, injects a key code directly instead of waiting for the next authentic signal.

Parameters:

keyCode (str) – The key code to inject as a string starting with 0x.

Raises:

RuntimeError if this signal listener does not allow injecting key codes. See signaListenerConfig.

describeProtocol()

Returns a string describing the protocol of this signal listener in a human-readable form (in English).

Useful for debugging or for log output.

Returns:

The description.

Return type:

str

Constants and defaults

gpiosvr.signal.TOLERANCE_PERCENTAGE_DEFAULT = 45
gpiosvr.signal.DEBOUNCE_μS_DEFAULT = 500
class gpiosvr.signal.BitOrder(value)

The bit order to use when translating the logical bits of a signal to a hex code.

LSB_FIRST = 0

The least significant bit (LSB) shall go first in the hex code.

MSB_FIRST = 1

The most significant bit (MSB) shall go first in the hex code.

gpiosvr.signal.BIT_ORDER_DEFAULT = BitOrder.MSB_FIRST
gpiosvr.signal.CODE_START_BIT_DEFAULT = 'GPIO_LEVEL'
gpiosvr.signal.MAX_BASE_PULSE_COUNT = 100

Helper classes

class gpiosvr.signal.PulseHelper
static paramsToBitCount(basePulse_μs, minEdgeCount, basePulseCount)

Calculates the number of logical bits that will be decoded from signals with the given parameters.

The parameters have the same meanings as with __init__().

Returns:

The number of logical bits.

Return type:

int

static sequenceToBasePulseCount(pulses_μs, basePulse_μs, tolerancePercentage=0)

For a sequence of actual pulse lengths, calculates the number of contained bases pulses. Applicable only if a fixed base pulse length is given.

Parameters:

pulses_μs (list[int] | tuple[int]) – The sequence of actual pulse lengths in microseconds to be evaluated.

The parameters basePulse_μs, tolerancePercentage have the same meanings as with __init__().

Returns:

The total number of base pulses contained in the pulse sequence.

Return type:

int

static sequenceToHexCodes(pulses_μs, basePulse_μs=300000, tolerancePercentage=0, bitCount=1, bitOrder=BitOrder.MSB_FIRST, pulseStartLevel=1, forceSingleCode=False, debugOutput: StringIO | None = None)

Produces a hex code for an actual pulse sequence, first converting the pulse sequence to logical bits.

Parameters:

  • pulses_μs (list[int] | tuple[int]) – The sequence of actual pulse lengths in microseconds to be evaluated.

  • pulseStartLevel (int) – The actual GPIO level of the first edge in the pulse sequence. May be either 0 or 1.

  • forceSingleCode (bool) – If True, returns at most one hex code. If False, returns as many hex codes as are contained in the pulse sequence. For instance, if the pulse sequence is decoded to 12 logical bits and bitCount is 5, the number of produced hex counts will be math.ceil(12 / 5) = 3. Defaults to False.

  • debugOutput (StringIO, optional) – A string buffer for appending debug messages. Useful for analysing unexpected hex codes. Defaults to None.

The parameters basePulse_μs, tolerancePercentage, bitCount and bitOrder have the same meanings as with __init__().

Returns:

The produced hex code as a str starting with 0x, all lowercase.

Return type:

list[str]