MatchCalls

Matching functions are used by the Frame Router to alert services, connections, and other features to frames which are relevant to them.

When a service or other feature (such as an individual connection) needs to receive frames from the frame router, it registers a MatchCall NamedTuple which consists of a matching function and a notifier.

Notifiers and matchers can be pure functions, closures, or bound methods, allowing you to construct services that track state and react to frames sent over the air.

Matching functions

Matching functions take a frame and an interface it was heard on, and return a boolean that indicates whether the frame matches. For example, we might construct a matcher called has_our_dest_address like so:

def has_our_dest_name(frame, interface):
    """
    Checks if a frame has our station name as its destination.
    """
    return interface.station.name == frame.route.dest.address.name

Notifiers

If a frame matches a matching function, its notifier is then called. Notifiers also take an interface and a frame as arguments, but they are not expected to return anything, and any returned information will be ignored. Here's an example notifier function that just prints whatever the matching frame was:

def print_relevant_frame(frame, interface):
    """
    Prints a frame and what interface it was received on.
    """
    print(f"On {interface.name}, received: {frame}")

Reference

Several matching functions and utilities are included with pax25. You can find their reference information below:

Matching utilities. Used for composing routing rules.

MatchCall

Bases: NamedTuple

A pair of functions-- one for matching, and one for processing a frame.

Used by the FrameRouter-- if a matcher function matches, calls the notify function.

Notifiers must be synchronous. If they need to perform an async action, have them use an async queue to defer the result. This is to prevent blocking the entire station on an async background action.

Source code in pax25/ax25/matchers.py

class MatchCall(NamedTuple):
    """
    A pair of functions-- one for matching, and one for processing a frame.

    Used by the FrameRouter-- if a matcher function matches, calls the notify function.

    Notifiers must be synchronous. If they need to perform an async action, have them
    use an async queue to defer the result. This is to prevent blocking the entire
    station on an async background action.
    """

    matcher: Matcher
    notify: Notifier

check_all(*checks: Matcher) -> Matcher

Returns True if all specified matcher functions return True.

Note

Passing no matchers will always return True.

Tip

Put faster-matching functions near the beginning of the list. Matching functions which iterate over lists will be used on every frame, potentially slowing processing if they are invoked when a faster matcher would otherwise decide the matching status.

Source code in pax25/ax25/matchers.py

def check_all(*checks: Matcher) -> Matcher:
    """
    Returns True if all specified matcher functions return True.

    !!! note

        Passing no matchers will always return True.

    !!! tip

        Put faster-matching functions near the beginning of the list. Matching
        functions which iterate over lists will be used on every frame, potentially
        slowing processing if they are invoked when a faster matcher would otherwise
        decide the matching status.
    """

    def wrapped(frame: Frame, interface: Interface) -> bool:
        """Bound wrapper."""
        return all(check(frame, interface) for check in checks)

    return wrapped

check_any(*checks: Matcher) -> Matcher

Returns True if any of the specified matcher functions return True.

Note

Passing no matchers will always return True.

Tip

Put faster-matching functions near the beginning of the list. Matching functions which iterate over lists will be used on every frame, potentially slowing processing if they are invoked when a faster matcher would otherwise decide the matching status.

Source code in pax25/ax25/matchers.py

def check_any(*checks: Matcher) -> Matcher:
    """
    Returns True if any of the specified matcher functions return True.

    !!! note

        Passing no matchers will always return True.

    !!! tip

        Put faster-matching functions near the beginning of the list. Matching
        functions which iterate over lists will be used on every frame, potentially
        slowing processing if they are invoked when a faster matcher would otherwise
        decide the matching status.
    """

    def wrapped(frame: Frame, interface: Interface) -> bool:
        """Bound wrapper."""
        return any(check(frame, interface) for check in checks)

    return wrapped

has_dest_address(address: Address) -> Matcher

Check if the given frame has the specific destination address.

Source code in pax25/ax25/matchers.py

def has_dest_address(address: Address) -> Matcher:
    """
    Check if the given frame has the specific destination address.
    """

    def wrapped(frame: Frame, _interface: Interface) -> bool:
        """Bound wrapper."""
        return frame.route.dest.address == address

    return wrapped

has_src_address(address: Address) -> Matcher

Check if the given frame has the specific destination address.

Source code in pax25/ax25/matchers.py

def has_src_address(address: Address) -> Matcher:
    """
    Check if the given frame has the specific destination address.
    """

    def wrapped(frame: Frame, _interface: Interface) -> bool:
        return frame.route.src.address == address

    return wrapped

has_these_digipeaters(digipeaters: tuple[Address, ...]) -> Matcher

Checks if a connection has a specific list of intermediate digipeaters.

Source code in pax25/ax25/matchers.py

def has_these_digipeaters(digipeaters: tuple[Address, ...]) -> Matcher:
    """
    Checks if a connection has a specific list of intermediate digipeaters.
    """

    def wrapped(frame: Frame, _interface: Interface) -> bool:
        """Bound wrapper."""
        return digipeaters == tuple(
            digipeater.address for digipeater in frame.route.digipeaters
        )

    return wrapped

is_connection_frame_for(src: Address, dest: Address, digipeaters: tuple[Address, ...]) -> Matcher

Checks if a frame belongs to a specific connection defined by the src, dest, and digipeater path.

Source code in pax25/ax25/matchers.py

def is_connection_frame_for(
    src: Address, dest: Address, digipeaters: tuple[Address, ...]
) -> Matcher:
    """
    Checks if a frame belongs to a specific connection defined by the src, dest, and
    digipeater path.
    """
    return check_all(
        has_src_address(src),
        has_dest_address(dest),
        has_these_digipeaters(digipeaters),
        repeats_completed,
        check_any(
            is_s_frame,
            is_i_frame,
        ),
    )

is_i_frame(frame: Frame, _interface: Interface) -> bool

Checks if the frame is an I-frame.

Source code in pax25/ax25/matchers.py

def is_i_frame(frame: Frame, _interface: Interface) -> bool:
    """
    Checks if the frame is an I-frame.
    """
    return frame.control.type == FrameType.INFORMATIONAL

is_s_frame(frame: Frame, _interface: Interface) -> bool

Checks if the frame is an S frame.

Source code in pax25/ax25/matchers.py

def is_s_frame(frame: Frame, _interface: Interface) -> bool:
    """
    Checks if the frame is an S frame.
    """
    return frame.control.type == FrameType.SUPERVISORY

is_u_frame(frame: Frame, _interface: Interface) -> bool

Checks if the frame is a U frame.

Source code in pax25/ax25/matchers.py

def is_u_frame(frame: Frame, _interface: Interface) -> bool:
    """
    Checks if the frame is a U frame.
    """
    return frame.control.type == FrameType.UNNUMBERED

needs_repeat_from(address: Address) -> Matcher

Checks if the given frame needs repeating by the given address.

Only does so if it's the next repeater in the set. Otherwise, ignores.

Source code in pax25/ax25/matchers.py

def needs_repeat_from(address: Address) -> Matcher:
    """
    Checks if the given frame needs repeating by the given address.

    Only does so if it's the next repeater in the set. Otherwise, ignores.
    """

    def wrapped(frame: Frame, _interface: Interface) -> bool:
        """Bound wrapper."""
        return should_repeat_for(address, frame)

    return wrapped

on_gateway(_frame: Frame, interface: Interface) -> bool

Checks if a frame was heard on a gateway interface.

Source code in pax25/ax25/matchers.py

def on_gateway(_frame: Frame, interface: Interface) -> bool:
    """
    Checks if a frame was heard on a gateway interface.
    """
    return interface.gateway

repeats_completed(frame: Frame, _interface: Interface) -> bool

Checks that a frame has finished making its way through the repeat chain.

Source code in pax25/ax25/matchers.py

def repeats_completed(frame: Frame, _interface: Interface) -> bool:
    """
    Checks that a frame has finished making its way through the repeat chain.
    """
    if not frame.route.digipeaters:
        return True
    return all(digi.command_or_repeated for digi in frame.route.digipeaters)