Skip to content

Applications

The application class is used to define applications that can be run on a station. Applications are registered with a station, and can be run by any connected client.

Registering Applications

Registering an application with the station is done using the add_app method on the connection service. This method takes three arguments: the application to register, a list of interfaces to register the application with, and a dictionary of settings for the application.

In addition to this, there are also some optional arguments that can be passed to the register_app method. Here's some example of the method in action:

Basic Registration

from pax25.contrib import Echo

...

# Basic invocation. It will register the Echo application with the station,
# and make it available on the file interface. It will respond to connections
# to the station's default name and its next available SSID. If this was the
# first application registered, and our callsign was N0CALL, it would be
# available as N0CALL, or NOCALL-0, which are aliases.
station.connection.add_app(Echo, ["file"], settings={})

Specifying the Station Name

from pax25.contrib import Echo

...

# In this case, we use a different station name, so this would be MYECHO.
station.connection.add_app(Echo, ["file"], station_name="MYECHO")
# If we ran this again, it would create a new copy of the echo app, and
# bump the SSID. The following app would be available on MYECHO-1.
station.connection.add_app(Echo, ["file"], station_name="MYECHO")

Specifying the SSID

from pax25.contrib import Echo

...

# In this case, we set both the statin name and the SSID. This would be MYECHO-5.
station.connection.add_app(Echo, ["file"], station_name="MYECHO", ssid=5, settings={})
# !! This will raise an error, as the SSID is already taken!
station.connection.add_app(Echo, ["file"], station_name="MYECHO", ssid=5, settings={})

Specifying multiple interfaces

from pax25.contrib import Echo

...

# Here's a single interface registration, on the file interface.
station.connection.add_app(Echo, ["file"], settings={})
# Here's a second registration on two interfaces. NOTE: because we
# already have an application on the file interface, the callsign on the
# file interface will be N0CALL-1 while the callsign on the TCP interface
# will be N0CALL-0!
station.connection.add_app(Echo, ["file", "tcp"], settings={})

Application Lifecycle

An application is instantiated when it is registered with a station. The application is then run when a client connects to the station and requests to run the application.

When the application is first registered, the setup method is called. This method is used to set up any state scaffolding that the application needs. For example, if the application needs to keep track of state for connected users, it will usually create a dictionary for storing that state on itself. See the HighLow example in the tutorial.

When a client connects to the station and requests to run the application, the on_startup method is called. This method is used to set up any state that the application needs to run for the duration of that connection. For example, if the application needs to keep track of state for the connected user, it will usually register the state in the dictionary it created in setup.

When the client disconnects from the station, the on_shutdown method is called. This method is used to clean up any state that the application needs to clean up. This might mean deleting the data it created in on_startup.

Utilities for creating applications

To make building applications easier, some included building blocks are provided-- a command router for routing commands the user types, and a help system for providing help. We strongly recommend use of these features to simplify your development experience, but they are not required.

pax25.applications.application.Application

The Application class. You should inherit from this class to create your own custom pax25 apps.

Source code in pax25/applications/application.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class Application(BaseApplication[S]):
    """
    The Application class. You should inherit from this class to create your own custom
    pax25 apps.
    """

    def __init__(self, *, name: str, station: "Station", settings: S):
        self.name = name
        self.settings = settings
        self.connection_state_table: BasicStateTable = {}
        self.station = station
        self.setup()

    def setup(self) -> None:
        """
        Perform any initial state configuration for your application in this
        function.
        """

    def is_admin(self, connection: "Connection") -> bool:
        """
        Check if the current user is an admin.
        """
        return connection.is_admin and connection.first_party.name == self.station.name

    def on_connect(self, connection: "Connection") -> None:
        """
        Called when a new connection is established.
        """
        self.connection_state_table[connection] = {"command": b""}
        self.on_startup(connection)

    def on_startup(self, connection: "Connection") -> None:
        """
        Run right after a new connection is established. You can use this function to
        do any initial state configuration and/or send a welcome message.
        """

    def on_disconnect(self, connection: "Connection") -> None:
        """
        Run when a connection is being dropped.
        """
        self.on_shutdown(connection)
        del self.connection_state_table[connection]

    def on_shutdown(self, connection: "Connection") -> None:
        """
        Called when a connection is being disconnected. Perform any cleanup here.
        """

    def on_message(self, connection: "Connection", message: str) -> None:
        """
        Called when a message is received. By default, this is called by
        on_bytes when it detects a carriage return has been sent.
        """

    def on_bytes(
        self,
        connection: "Connection",
        bytes_received: bytes,
    ) -> None:
        """
        Called when bytes are received from a connection for this application. You
        usually don't want to call this directly, but you might need to if you need
        to control how bytes sent from the client are handled.
        """
        # Not sure how often we'll receive packets with carriage returns in the
        # middle of them, but for most applications that should indicate the
        # end of one command and the start of another, so we break them up here.
        for raw_int in bytes_received:
            # Iterating over bytes produces ints.
            byte = raw_int.to_bytes(1, sys.byteorder)
            try:
                self.connection_state_table[connection]["command"]
            except KeyError:
                # Disconnected between our previous processed byte and our current one.
                return
            match byte:
                case b"\r":
                    current_message = (
                        self.connection_state_table[connection]["command"] + byte
                    ).decode("utf-8")
                    # Clear the command before sending the message in case there's an
                    # exception.
                    self.connection_state_table[connection]["command"] = b""
                    self.on_message(
                        connection,
                        # Remove trailing newline.
                        current_message[:-1],
                    )
                # Backspace
                case b"\x7f":
                    current_string = self.connection_state_table[connection][
                        "command"
                    ].decode("utf-8", errors="backslashreplace")
                    self.connection_state_table[connection]["command"] = current_string[
                        :-1
                    ].encode("utf-8")
                case _:
                    self.connection_state_table[connection]["command"] += byte

setup() -> None

Perform any initial state configuration for your application in this function.

Source code in pax25/applications/application.py 
28
29
30
31
32
def setup(self) -> None:
    """
    Perform any initial state configuration for your application in this
    function.
    """

is_admin(connection: Connection) -> bool

Check if the current user is an admin.

Source code in pax25/applications/application.py 
34
35
36
37
38
def is_admin(self, connection: "Connection") -> bool:
    """
    Check if the current user is an admin.
    """
    return connection.is_admin and connection.first_party.name == self.station.name

on_startup(connection: Connection) -> None

Run right after a new connection is established. You can use this function to do any initial state configuration and/or send a welcome message.

Source code in pax25/applications/application.py 
47
48
49
50
51
def on_startup(self, connection: "Connection") -> None:
    """
    Run right after a new connection is established. You can use this function to
    do any initial state configuration and/or send a welcome message.
    """

on_shutdown(connection: Connection) -> None

Called when a connection is being disconnected. Perform any cleanup here.

Source code in pax25/applications/application.py 
60
61
62
63
def on_shutdown(self, connection: "Connection") -> None:
    """
    Called when a connection is being disconnected. Perform any cleanup here.
    """

on_message(connection: Connection, message: str) -> None

Called when a message is received. By default, this is called by on_bytes when it detects a carriage return has been sent.

Source code in pax25/applications/application.py 
65
66
67
68
69
def on_message(self, connection: "Connection", message: str) -> None:
    """
    Called when a message is received. By default, this is called by
    on_bytes when it detects a carriage return has been sent.
    """

on_bytes(connection: Connection, bytes_received: bytes) -> None

Called when bytes are received from a connection for this application. You usually don't want to call this directly, but you might need to if you need to control how bytes sent from the client are handled.

Source code in pax25/applications/application.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def on_bytes(
    self,
    connection: "Connection",
    bytes_received: bytes,
) -> None:
    """
    Called when bytes are received from a connection for this application. You
    usually don't want to call this directly, but you might need to if you need
    to control how bytes sent from the client are handled.
    """
    # Not sure how often we'll receive packets with carriage returns in the
    # middle of them, but for most applications that should indicate the
    # end of one command and the start of another, so we break them up here.
    for raw_int in bytes_received:
        # Iterating over bytes produces ints.
        byte = raw_int.to_bytes(1, sys.byteorder)
        try:
            self.connection_state_table[connection]["command"]
        except KeyError:
            # Disconnected between our previous processed byte and our current one.
            return
        match byte:
            case b"\r":
                current_message = (
                    self.connection_state_table[connection]["command"] + byte
                ).decode("utf-8")
                # Clear the command before sending the message in case there's an
                # exception.
                self.connection_state_table[connection]["command"] = b""
                self.on_message(
                    connection,
                    # Remove trailing newline.
                    current_message[:-1],
                )
            # Backspace
            case b"\x7f":
                current_string = self.connection_state_table[connection][
                    "command"
                ].decode("utf-8", errors="backslashreplace")
                self.connection_state_table[connection]["command"] = current_string[
                    :-1
                ].encode("utf-8")
            case _:
                self.connection_state_table[connection]["command"] += byte